aios/active-runtime/app/isolated_writer_authz.py

"""Fail-closed operation authorization for an isolated Git writer."""
from __future__ import annotations

import ipaddress
import os
import re
import subprocess
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from isolated_writer_token_registry import TokenRegistry


ROUTE_OPERATIONS = {
    ("GET", "/api/runtime-status"): "runtime.status.read",
    ("GET", "/api/baseline"): "baseline.read",
    ("GET", "/api/patches"): "patches.read",
    ("POST", "/api/commit-file"): "commit_file.atomic",
}
OID_RE = re.compile(r"[0-9a-f]{40}")


class AuthorizationError(PermissionError):
    def __init__(self, code: str, status: int = 403):
        super().__init__(code)
        self.code = code
        self.status = status


@dataclass(frozen=True)
class AuthorizedGrant:
    record: dict[str, Any]
    operation: str
    use_count: int

    @property
    def token_id(self) -> str:
        return str(self.record["token_id"])

    @property
    def task_id(self) -> str:
        return str(self.record["task_id"])


def operation_for(method: str, path: str) -> str:
    operation = ROUTE_OPERATIONS.get((method.upper(), path))
    if not operation:
        raise AuthorizationError("isolated_writer_operation_forbidden")
    return operation


def _parse_time(value: Any) -> datetime:
    try:
        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except (TypeError, ValueError):
        raise AuthorizationError("writer_token_time_invalid") from None
    if parsed.tzinfo is None:
        raise AuthorizationError("writer_token_time_invalid")
    return parsed.astimezone(timezone.utc)


def _git_is_ancestor(workspace: Path, base: str, branch: str) -> bool:
    result = subprocess.run(
        ["git", "merge-base", "--is-ancestor", base, branch],
        cwd=workspace,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    return result.returncode == 0


def _validate_grant(
    grant: dict[str, Any],
    *,
    audience: str,
    source_ip: str,
    operation: str,
    branch: str | None,
    path: str | None,
    workspace: Path,
    current_time: datetime,
) -> None:
    if int(grant.get("schema_version", 0)) != 2:
        raise AuthorizationError("writer_token_policy_v2_required", 401)
    if str(grant.get("audience") or "") != audience:
        raise AuthorizationError("writer_token_audience_rejected")
    try:
        peer = ipaddress.ip_address(source_ip)
        networks = [ipaddress.ip_network(str(value), strict=True) for value in grant.get("source_cidrs", [])]
    except ValueError:
        raise AuthorizationError("writer_token_source_rejected") from None
    if not networks or not any(peer in network for network in networks):
        raise AuthorizationError("writer_token_source_rejected")
    created = _parse_time(grant.get("created_at"))
    not_before = _parse_time(grant.get("not_before"))
    expires = _parse_time(grant.get("expires_at"))
    if expires <= not_before or (expires - created).total_seconds() > 1800:
        raise AuthorizationError("writer_token_lifetime_rejected")
    if current_time < not_before:
        raise AuthorizationError("writer_token_not_yet_valid", 401)
    if current_time >= expires:
        raise AuthorizationError("writer_token_expired", 401)
    if bool(grant.get("revoked")):
        raise AuthorizationError("writer_token_revoked", 401)
    operations = grant.get("allowed_operations")
    if not isinstance(operations, list) or operation not in operations:
        raise AuthorizationError("writer_token_operation_rejected")
    expected_branch = str(grant.get("allowed_branch") or "")
    configured_branch = os.getenv("AIOS_GITOPS_ALLOWED_BRANCH", "")
    if expected_branch != configured_branch:
        raise AuthorizationError("writer_token_branch_rejected")
    if branch is not None and branch != expected_branch:
        raise AuthorizationError("writer_token_branch_rejected")
    if path is not None:
        paths = grant.get("allowed_paths")
        if not isinstance(paths, list) or path not in paths:
            raise AuthorizationError("writer_token_path_rejected")
    base = str(grant.get("approved_base_oid") or "").lower()
    if not OID_RE.fullmatch(base):
        raise AuthorizationError("writer_token_base_rejected")
    if not _git_is_ancestor(workspace, base, expected_branch):
        raise AuthorizationError("writer_token_base_not_ancestor")


def authorize(
    registry: TokenRegistry,
    token: str,
    *,
    method: str,
    path: str,
    source_ip: str,
    workspace: Path,
    branch: str | None = None,
    target_path: str | None = None,
    current_time: datetime | None = None,
) -> AuthorizedGrant:
    if os.getenv("AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED", "") != "2":
        raise AuthorizationError("writer_token_policy_v2_not_required")
    if not token:
        raise AuthorizationError("unauthorized", 401)
    grant = registry.find(token)
    if grant is None:
        raise AuthorizationError("unauthorized", 401)
    operation = operation_for(method, path)
    _validate_grant(
        grant,
        audience=os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", ""),
        source_ip=source_ip,
        operation=operation,
        branch=branch,
        path=target_path,
        workspace=workspace,
        current_time=current_time or datetime.now(timezone.utc),
    )
    limits = grant.get("max_uses")
    if not isinstance(limits, dict):
        raise AuthorizationError("writer_token_use_limit_missing")
    try:
        used = registry.consume(str(grant["token_id"]), operation, int(limits[operation]))
    except (KeyError, TypeError, ValueError):
        raise AuthorizationError("writer_token_use_limit_missing") from None
    except PermissionError as exc:
        code = str(exc)
        raise AuthorizationError(code, 401 if code == "writer_token_revoked" else 403) from None
    return AuthorizedGrant(grant, operation, used)