74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
"""HTTP adapter for Owner-authorized task writer provisioning.
|
|
|
|
Run this module instead of aios_chatgpt_actions_api.py. All existing routes are
|
|
delegated unchanged; only POST /api/task-writer/provision-authorized is added.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import urllib.parse
|
|
|
|
import aios_chatgpt_actions_api as api
|
|
from authorized_task_writer_provisioning import (
|
|
AuthorizedTaskWriterProvisioner,
|
|
OwnerContext,
|
|
ProvisioningError,
|
|
)
|
|
|
|
|
|
_original_do_post = api.H.do_POST
|
|
_ENDPOINT = "/api/task-writer/provision-authorized"
|
|
|
|
|
|
def _owner_authorization(handler):
|
|
scope = os.getenv(
|
|
"AIOS_GITOPS_OWNER_PROVISION_SCOPE", "writer:provision"
|
|
)
|
|
auth = handler.need(scope)
|
|
if not auth:
|
|
return None
|
|
if auth.get("role") != "owner":
|
|
handler.sendj({"error": "OWNER_AUTHORIZATION_REQUIRED"}, 403)
|
|
return None
|
|
identity = str(auth.get("account") or "").strip()
|
|
if not identity:
|
|
handler.sendj({"error": "OWNER_IDENTITY_REQUIRED"}, 403)
|
|
return None
|
|
return auth
|
|
|
|
|
|
def _authorized_do_post(self):
|
|
path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
|
|
if path != _ENDPOINT:
|
|
return _original_do_post(self)
|
|
auth = _owner_authorization(self)
|
|
if not auth:
|
|
return
|
|
data = self.body()
|
|
owner_reference = str(data.get("owner_authorization_reference") or "")
|
|
try:
|
|
result = AuthorizedTaskWriterProvisioner().provision(
|
|
data,
|
|
OwnerContext(
|
|
identity=str(auth["account"]),
|
|
role=str(auth["role"]),
|
|
authorization_reference=owner_reference,
|
|
),
|
|
)
|
|
return self.sendj(result)
|
|
except ProvisioningError as exc:
|
|
body = {"error": exc.code}
|
|
if exc.details:
|
|
body["details"] = exc.details
|
|
return self.sendj(body, exc.status)
|
|
except Exception:
|
|
# Do not expose tool output, paths containing credentials, or traceback data.
|
|
return self.sendj({"error": "AUTHORIZED_WRITER_BRIDGE_INTERNAL_ERROR"}, 500)
|
|
|
|
|
|
api.H.do_POST = _authorized_do_post
|
|
|
|
|
|
if __name__ == "__main__":
|
|
api.main()
|