aios/active-runtime/app/aios_isolated_writer_api.py

"""Task-scoped hardened wrapper around the existing Action API."""
from __future__ import annotations

import json
import os
import subprocess
from pathlib import Path

import aios_chatgpt_actions_api as api
import aios_isolated_writer_policy as policy

_reject = api.reject_non_read_http
_write_file = api.write_file
_sync_branch = api.sync_branch


def reject_non_read_http(handler, method: str, path: str) -> bool:
    if _reject(handler, method, path):
        return True
    try:
        policy.validate_endpoint(method, path, api.WS, api.BARE_REPO)
        return False
    except PermissionError as exc:
        handler.sendj({"error": str(exc)}, 403)
        return True


def checkout(branch, base="main", create=False):
    api.ensure_git()
    target = api.branch_name(branch)
    policy.validate_branch_request({"branch": target, "base": base})
    exists = api.sh(["rev-parse", "--verify", target])["code"] == 0
    if exists:
        if api.sh(["rev-parse", "--abbrev-ref", "HEAD"], True)["stdout"] != target:
            api.sh(["checkout", target], True)
    elif create:
        api.sh(["checkout", "-b", target, base], True)
    else:
        raise PermissionError("isolated_writer_branch_missing")
    policy.require_service_invariants(api.WS, api.BARE_REPO)
    return target


def write_file(path, content):
    policy.validate_path(path, api.WS)
    return _write_file(path, content)


def sync_branch(branch: str):
    policy.validate_sync_branch(branch)
    local = _sync_branch(branch)
    remote = os.getenv("AIOS_GITOPS_GITEA_PUSH_REMOTE", "").strip()
    if remote != "gitea-http":
        raise PermissionError("isolated_writer_gitea_remote_rejected")
    head = api.sh(["rev-parse", "HEAD"], True)["stdout"]
    pushed = subprocess.run(
        ["git", "push", "--porcelain", remote, f"HEAD:refs/heads/{branch}"],
        cwd=api.WS,
        text=True,
        capture_output=True,
    )
    if pushed.returncode != 0:
        raise RuntimeError("isolated_writer_gitea_push_failed")
    verified = subprocess.run(
        ["git", "ls-remote", "--heads", remote, f"refs/heads/{branch}"],
        cwd=api.WS,
        text=True,
        capture_output=True,
    )
    if verified.returncode != 0:
        raise RuntimeError("isolated_writer_gitea_verification_failed")
    fields = verified.stdout.strip().split()
    if len(fields) != 2 or fields[0] != head or fields[1] != f"refs/heads/{branch}":
        raise RuntimeError("isolated_writer_gitea_oid_mismatch")
    return {
        "status": "verified",
        "target": "authoritative-bare+gitea",
        "branch": branch,
        "oid": head,
        "local": local.get("status"),
    }


def commit(paths, msg):
    policy.validate_commit_paths(paths, api.WS)
    api.ensure_git()
    selected = [api.relpath(p) for p in paths] if paths else ["."]
    api.sh(["add", *selected], True)
    policy.validate_staged_paths(api.WS)
    branch = api.sh(["rev-parse", "--abbrev-ref", "HEAD"], True)["stdout"]
    if not api.sh(["status", "--short"])["stdout"]:
        return {"status": "no_changes", "branch": branch, "sync": sync_branch(branch)}
    out = api.sh(["commit", "-m", msg], True)
    branch = api.sh(["rev-parse", "--abbrev-ref", "HEAD"], True)["stdout"]
    return {
        "status": "committed",
        "branch": branch,
        "rev": api.sh(["rev-parse", "--short", "HEAD"], True)["stdout"],
        "git": out,
        "sync": sync_branch(branch),
    }


def audit(actor, action, target, payload):
    policy.append_audit_record(
        {
            "at": api.now(),
            "actor": actor,
            "action": action,
            "target": target,
            "payload": api.sanitize(payload),
        }
    )


def _validate_request(self, data, validate):
    try:
        validate(data)
        return True
    except PermissionError as exc:
        self.sendj({"error": str(exc)}, 403)
        return False


def _checked_branch(self, data):
    if not _validate_request(self, data, policy.validate_branch_request):
        return
    return _branch(self, data)


def _checked_file(self, data):
    data = dict(data)
    data.setdefault("branch", policy.allowed_branch())
    if not _validate_request(self, data, policy.validate_branch_request):
        return
    if not _validate_request(
        self,
        {"path": data.get("path", "")},
        lambda item: policy.validate_path(item["path"], api.WS),
    ):
        return
    return _file(self, data)


def _checked_commit(self, data):
    data = dict(data)
    data.setdefault("branch", policy.allowed_branch())
    if not _validate_request(self, data, policy.validate_branch_request):
        return
    return _commit(self, data)


def _checked_patch(self, data):
    data = dict(data)
    data["branch"] = policy.allowed_branch()
    if not _validate_request(self, data, policy.validate_branch_request):
        return
    return _patch(self, data)


_branch, _file, _commit, _patch = api.H.branch, api.H.file, api.H.commit, api.H.patch
api.reject_non_read_http = reject_non_read_http
api.checkout, api.write_file, api.sync_branch, api.commit, api.audit = (
    checkout,
    write_file,
    sync_branch,
    commit,
    audit,
)
api.H.branch, api.H.file, api.H.commit, api.H.patch = (
    _checked_branch,
    _checked_file,
    _checked_commit,
    _checked_patch,
)

if __name__ == "__main__":
    policy.require_service_invariants(api.WS, api.BARE_REPO)
    api.ThreadingHTTPServer((api.HOST, api.PORT), api.H).serve_forever()