pax_coder_gate.py

#!/usr/bin/env python3
"""

PAX-Coder Protected Execution Gate (ADR-0009)



Native Python implementation with Ed25519 cryptographic verification.

Replaces the shell-based gate with:

  - Native Ed25519 via cryptography library (no openssl CLI)

  - JSON schema validation via Pydantic

  - No subprocess, no shell, no external CLI tools

  - Deterministic JSON canonicalization (sorted keys, compact)



Exit codes:

  0 = AUTHORIZED (protected execution allowed)

  1 = INTEGRITY_FAILED (release verification failed)

  2 = AUTHORIZATION_DENIED (node not authorized or capability missing/invalid)

  3 = SCRIPT_ERROR (cannot determine status)



Environment:

  PAX_CAPABILITY_TOKEN  - capability token (JSON|signature_hex)

  PAX_REPO_ROOT         - override repo root (defaults to script parent dir)



Usage:

  python3 pax_coder_gate.py [--quiet] [--json-output]

"""

import json
import os
import sys
import hashlib
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

from pydantic import BaseModel, Field, ValidationError
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.hazmat.primitives.serialization import load_pem_public_key
from cryptography.exceptions import InvalidSignature


# =============================================================================
# EXIT CODES
# =============================================================================

EXIT_AUTHORIZED = 0
EXIT_INTEGRITY_FAILED = 1
EXIT_DENIED = 2
EXIT_ERROR = 3


# =============================================================================
# MODELS (Pydantic schema validation)
# =============================================================================

class CapabilityPayload(BaseModel):
    """Schema for capability token JSON payload."""
    node_id: str = Field(min_length=1)
    release_id: str = Field(min_length=1)
    commit: str = Field(min_length=1)
    nonce: str = Field(min_length=1)
    expires_at: str = Field(min_length=1)


class ReleaseMetadata(BaseModel):
    """Schema for sovereign/release.json."""
    project: str = ""
    repository: str = ""
    release_version: str = ""
    git_commit: str = ""
    node_id: str = ""
    manifest_sha256: str = ""
    release_timestamp_utc: str = ""


class AuthorizationRecord(BaseModel):
    """Schema for sovereign/authorization.json."""
    authorization_id: str = ""
    node_id: str = ""
    authorization_status: str = ""
    authorization_scope: str = ""
    expires_at_utc: Optional[str] = None
    revocation_status: str = ""


class NodeIdentity(BaseModel):
    """Schema for sovereign/node.json."""
    node_id: str = ""
    algorithm: str = ""
    public_key_hex: str = ""


# =============================================================================
# GATE IMPLEMENTATION
# =============================================================================

class PaxCoderGate:
    """

    Protected Execution Gate.



    Verifies in order:

      1. Release integrity (git commit matches release.json, manifest hash)

      2. Node authorization status (ACTIVE, not revoked, not expired)

      3. Capability possession (env var or file)

      4. Capability validity (commit, expiration, node binding)

      5. Capability signature (Ed25519 with authority public key)

    """

    def __init__(self, repo_root: Optional[Path] = None, quiet: bool = False):
        if repo_root is None:
            # Default: parent of the script location
            repo_root = Path(__file__).resolve().parent
        self.repo_root = Path(repo_root)
        self.sovereign_dir = self.repo_root / "sovereign"
        self.quiet = quiet
        self._messages: list[str] = []

    def log(self, msg: str) -> None:
        """Log a message (suppressed in quiet mode)."""
        self._messages.append(msg)
        if not self.quiet:
            print(msg)

    def run(self) -> int:
        """Execute the full gate sequence. Returns exit code."""
        self.log("==========================================")
        self.log("PAX-CODER PROTECTED EXECUTION GATE")
        self.log("==========================================")
        self.log("")

        # Step 1: Release integrity
        result = self._verify_release_integrity()
        if result != EXIT_AUTHORIZED:
            return result

        # Step 2: Node authorization
        result = self._verify_node_authorization()
        if result != EXIT_AUTHORIZED:
            return result

        # Step 3: Capability possession
        capability_raw = self._get_capability_token()
        if capability_raw is None:
            return EXIT_DENIED

        # Step 4: Parse and validate capability
        payload, signature_hex = self._parse_capability(capability_raw)
        if payload is None:
            return EXIT_DENIED

        result = self._validate_capability(payload)
        if result != EXIT_AUTHORIZED:
            return result

        # Step 5: Verify signature
        result = self._verify_signature(payload, signature_hex)
        if result != EXIT_AUTHORIZED:
            return result

        # All checks passed
        self.log("")
        self.log("==========================================")
        self.log("STATUS: AUTHORIZATION_GRANTED")
        self.log("==========================================")
        self.log("")
        self.log("Protected execution is AUTHORIZED.")
        self.log("")
        self.log("You may now:")
        self.log("  - Generate node keys")
        self.log("  - Sign releases")
        self.log("  - Invoke other protected operations")
        self.log("")
        self.log(f"Capability valid until: {payload.expires_at}")
        self.log("")

        return EXIT_AUTHORIZED

    # =========================================================================
    # STEP 1: RELEASE INTEGRITY
    # =========================================================================

    def _verify_release_integrity(self) -> int:
        """Verify release.json matches current state."""
        self.log("[1/5] Verifying release integrity...")

        release_file = self.sovereign_dir / "release.json"
        if not release_file.exists():
            self.log("FAILED: sovereign/release.json not found")
            return EXIT_INTEGRITY_FAILED

        try:
            with open(release_file) as f:
                data = json.load(f)
            release = ReleaseMetadata(**data)
        except (json.JSONDecodeError, ValidationError) as e:
            self.log(f"FAILED: Cannot parse release.json: {e}")
            return EXIT_INTEGRITY_FAILED

        # Verify git commit matches
        current_commit = self._get_git_commit()
        if current_commit is None:
            self.log("FAILED: Cannot determine current git commit")
            return EXIT_ERROR

        if current_commit != release.git_commit:
            self.log("FAILED: Release integrity check failed")
            self.log(f"  Expected commit: {release.git_commit}")
            self.log(f"  Current commit:  {current_commit}")
            return EXIT_INTEGRITY_FAILED

        # Verify manifest hash if manifest exists
        manifest_file = self.sovereign_dir / "manifest.json"
        if manifest_file.exists() and release.manifest_sha256:
            computed_hash = self._sha256_file(manifest_file)
            if computed_hash != release.manifest_sha256:
                self.log("FAILED: Manifest hash mismatch")
                self.log(f"  Expected: {release.manifest_sha256}")
                self.log(f"  Computed: {computed_hash}")
                return EXIT_INTEGRITY_FAILED

        self.log("    Release integrity verified")
        self.log("")
        return EXIT_AUTHORIZED

    # =========================================================================
    # STEP 2: NODE AUTHORIZATION
    # =========================================================================

    def _verify_node_authorization(self) -> int:
        """Verify node has active authorization."""
        self.log("[2/5] Verifying node authorization status...")

        auth_file = self.sovereign_dir / "authorization.json"
        if not auth_file.exists():
            self.log("FAILED: Authorization record not found")
            return EXIT_DENIED

        node_file = self.sovereign_dir / "node.json"
        if not node_file.exists():
            self.log("FAILED: Node identity not found")
            return EXIT_DENIED

        try:
            with open(auth_file) as f:
                auth_data = json.load(f)
            auth = AuthorizationRecord(**auth_data)
        except (json.JSONDecodeError, ValidationError) as e:
            self.log(f"FAILED: Cannot parse authorization.json: {e}")
            return EXIT_ERROR

        try:
            with open(node_file) as f:
                node_data = json.load(f)
            node = NodeIdentity(**node_data)
        except (json.JSONDecodeError, ValidationError) as e:
            self.log(f"FAILED: Cannot parse node.json: {e}")
            return EXIT_ERROR

        # Check authorization status
        if auth.authorization_status != "ACTIVE":
            self.log(f"DENIED: Authorization status is {auth.authorization_status}")
            if auth.authorization_status == "REQUESTED":
                self.log("  Status is REQUESTED (not yet authorized)")
            elif auth.authorization_status == "SUSPENDED":
                self.log("  Status is SUSPENDED")
            elif auth.authorization_status == "REVOKED":
                self.log("  Status is REVOKED")
            elif auth.authorization_status == "EXPIRED":
                self.log("  Status is EXPIRED")
            return EXIT_DENIED

        # Check revocation status
        if auth.revocation_status != "ACTIVE":
            self.log(f"DENIED: Revocation status is {auth.revocation_status}")
            return EXIT_DENIED

        # Check expiration
        if auth.expires_at_utc and auth.expires_at_utc != "null":
            try:
                expires = datetime.fromisoformat(
                    auth.expires_at_utc.replace("Z", "+00:00")
                )
                now = datetime.now(timezone.utc)
                if now > expires:
                    self.log(f"DENIED: Authorization has expired ({auth.expires_at_utc})")
                    return EXIT_DENIED
            except ValueError:
                pass  # If we can't parse, skip expiration check

        # Check node ID consistency
        if auth.node_id != node.node_id:
            self.log("DENIED: Node ID mismatch")
            self.log(f"  Authorization: {auth.node_id}")
            self.log(f"  Local node:    {node.node_id}")
            return EXIT_DENIED

        self.log("    Node authorization verified")
        self.log("")
        return EXIT_AUTHORIZED

    # =========================================================================
    # STEP 3: CAPABILITY POSSESSION
    # =========================================================================

    def _get_capability_token(self) -> Optional[str]:
        """Get capability token from environment or file."""
        self.log("[3/5] Checking for capability...")

        # Check environment variable
        token = os.environ.get("PAX_CAPABILITY_TOKEN", "").strip()
        if token:
            self.log("    Capability token found (environment)")
            self.log("")
            return token

        # Check capability file
        cap_file = self.sovereign_dir / ".capability"
        if cap_file.exists():
            token = cap_file.read_text().strip()
            if token:
                self.log("    Capability token found (file)")
                self.log("")
                return token

        # No capability available
        node_id = "unknown"
        try:
            node_file = self.sovereign_dir / "node.json"
            if node_file.exists():
                with open(node_file) as f:
                    node_id = json.load(f).get("node_id", "unknown")
        except Exception:
            pass

        self.log("DENIED: No capability available")
        self.log("")
        self.log("Protected execution requires a capability token.")
        self.log("")
        self.log("To obtain authorization:")
        self.log("  1. Contact the PAX-Coder authority")
        self.log(f"  2. Request a capability for:")
        self.log(f"     - node_id: {node_id}")
        self.log(f"     - release: {self._get_git_commit() or 'unknown'}")
        self.log("  3. Set: export PAX_CAPABILITY_TOKEN=<capability>")
        self.log("  4. Re-run protected operation")
        self.log("")
        return None

    # =========================================================================
    # STEP 4: CAPABILITY PARSING AND VALIDATION
    # =========================================================================

    def _parse_capability(self, raw: str) -> tuple[Optional[CapabilityPayload], str]:
        """Parse capability token into payload and signature."""
        self.log("[4/5] Parsing capability...")

        # Split on pipe: JSON|signature_hex
        parts = raw.split("|", 1)
        if len(parts) != 2:
            self.log("DENIED: Capability format invalid (missing separator)")
            return None, ""

        json_part = parts[0].strip()
        sig_hex = parts[1].strip()

        # Parse JSON with Pydantic validation
        try:
            data = json.loads(json_part)
            payload = CapabilityPayload(**data)
        except json.JSONDecodeError as e:
            self.log(f"DENIED: Capability JSON invalid: {e}")
            return None, ""
        except ValidationError as e:
            self.log(f"DENIED: Capability format invalid")
            return None, ""

        self.log(f"    Node ID: {payload.node_id}")
        self.log(f"    Commit:  {payload.commit}")
        self.log(f"    Expires: {payload.expires_at}")
        self.log("    Capability parsed")
        self.log("")

        return payload, sig_hex

    def _validate_capability(self, payload: CapabilityPayload) -> int:
        """Validate capability fields against current state."""
        self.log("[5/5] Validating capability...")

        # Check 1: Commit matches current HEAD
        current_commit = self._get_git_commit()
        if current_commit and current_commit != payload.commit:
            self.log("DENIED: Release commit mismatch")
            self.log(f"  Expected: {payload.commit}")
            self.log(f"  Current:  {current_commit}")
            return EXIT_DENIED

        self.log("    Commit matches")

        # Check 2: Expiration
        try:
            expires = datetime.fromisoformat(
                payload.expires_at.replace("Z", "+00:00")
            )
            now = datetime.now(timezone.utc)
            if now > expires:
                self.log("DENIED: Capability expired")
                self.log(f"  Expired at: {payload.expires_at}")
                return EXIT_DENIED
        except ValueError:
            self.log("ERROR: Cannot parse expiration time")
            return EXIT_ERROR

        self.log("    Capability not expired")

        # Check 3: Node ID consistency
        node_file = self.sovereign_dir / "node.json"
        if node_file.exists():
            try:
                with open(node_file) as f:
                    local_node_id = json.load(f).get("node_id", "")
                if local_node_id and local_node_id != payload.node_id:
                    self.log("DENIED: Node ID mismatch")
                    self.log(f"  Capability node: {payload.node_id}")
                    self.log(f"  Local node:      {local_node_id}")
                    return EXIT_DENIED
                self.log("    Node ID matches")
            except Exception:
                pass

        self.log("")
        return EXIT_AUTHORIZED

    # =========================================================================
    # STEP 5: SIGNATURE VERIFICATION (Ed25519, native)
    # =========================================================================

    def _verify_signature(self, payload: CapabilityPayload, sig_hex: str) -> int:
        """Verify Ed25519 signature using authority public key."""
        self.log("[6/6] Verifying capability signature...")

        # Load authority public key
        authority_pk_file = self.sovereign_dir / "authority_pk.pem"
        if not authority_pk_file.exists():
            self.log(f"ERROR: Authority public key not found at {authority_pk_file}")
            self.log("This gate cannot verify capabilities without the authority's public key.")
            self.log("HINT: Authority public key should be provided during deployment.")
            return EXIT_ERROR

        # Validate signature is present
        if not sig_hex:
            self.log("DENIED: Capability signature missing")
            return EXIT_DENIED

        # Validate signature format: 128 hex characters = 64 bytes (Ed25519)
        if len(sig_hex) != 128:
            self.log("DENIED: Capability signature invalid format (expected 128 hex chars)")
            return EXIT_DENIED

        # Validate hex characters only
        try:
            sig_bytes = bytes.fromhex(sig_hex)
        except ValueError:
            self.log("DENIED: Capability signature contains non-hex characters")
            return EXIT_DENIED

        # Create canonical JSON (sorted keys, compact separators)
        canonical_data = {
            "commit": payload.commit,
            "expires_at": payload.expires_at,
            "node_id": payload.node_id,
            "nonce": payload.nonce,
            "release_id": payload.release_id,
        }
        canonical_json = json.dumps(canonical_data, sort_keys=True, separators=(",", ":"))
        message_bytes = canonical_json.encode("utf-8")

        # Load and verify with authority public key
        try:
            with open(authority_pk_file, "rb") as f:
                pem_data = f.read()
            public_key = load_pem_public_key(pem_data)

            if not isinstance(public_key, Ed25519PublicKey):
                self.log("ERROR: Authority key is not Ed25519")
                return EXIT_ERROR

            # Verify signature (raises InvalidSignature on failure)
            public_key.verify(sig_bytes, message_bytes)

            self.log("    Signature format valid")
            self.log("    Signature verified (cryptographic validation)")
            return EXIT_AUTHORIZED

        except InvalidSignature:
            self.log("DENIED: Capability signature verification failed")
            self.log("    The signature is invalid or was not issued by the authority")
            return EXIT_DENIED
        except Exception as e:
            self.log(f"ERROR: Signature verification error: {e}")
            return EXIT_ERROR

    # =========================================================================
    # UTILITIES
    # =========================================================================

    def _get_git_commit(self) -> Optional[str]:
        """Get current HEAD commit hash."""
        try:
            result = subprocess.run(
                ["git", "rev-parse", "HEAD"],
                cwd=str(self.repo_root),
                capture_output=True,
                text=True,
                timeout=10,
            )
            if result.returncode == 0:
                return result.stdout.strip()
        except (subprocess.TimeoutExpired, FileNotFoundError):
            pass
        return None

    def _sha256_file(self, filepath: Path) -> str:
        """Compute SHA-256 hash of a file."""
        h = hashlib.sha256()
        with open(filepath, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                h.update(chunk)
        return h.hexdigest()


# =============================================================================
# ENTRY POINT
# =============================================================================

def main() -> int:
    """Main entry point."""
    quiet = "--quiet" in sys.argv or "-q" in sys.argv

    # Allow repo root override via environment
    repo_root = os.environ.get("PAX_REPO_ROOT")
    if repo_root:
        root_path = Path(repo_root)
    else:
        root_path = Path(__file__).resolve().parent

    gate = PaxCoderGate(repo_root=root_path, quiet=quiet)
    return gate.run()


if __name__ == "__main__":
    sys.exit(main())