feat(gpt-edge): add owner-approved activation and rollback controller

2026-08-19 10:34:20 +08:00
parent 69a51d50c7
commit 3a02a25d79
3 changed files with 1037 additions and 0 deletions
@@ -0,0 +1,561 @@
+#!/usr/bin/env python3
+"""Owner-approved activation/rollback controller for the AIOS GPT HTTPS edge.
+
+This controller never creates approval, TLS, or ingress credentials. It consumes
+owner-managed inputs, applies one exact public-bind overlay, verifies the live
+edge, and provides an idempotent rollback path.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import socket
+import ssl
+import stat
+import subprocess
+import tempfile
+import time
+import urllib.request
+from typing import Any, Mapping
+
+APPROVAL_SCHEMA = "aios.gpt-https-edge.public-approval.v1"
+STATE_SCHEMA = "aios.gpt-https-edge.activation-state.v1"
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+SHA_RE = re.compile(r"^[0-9a-f]{64}$")
+APPROVAL_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$")
+DOMAIN_RE = re.compile(
+    r"(?=.{1,253}\Z)"
+    r"(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+"
+    r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?"
+)
+
+EDGE_UNIT = "aios-gpt-https-edge.service"
+INGRESS_UNIT = "aios-gpt-online-task-ingress.service"
+PUBLIC_HOST = "0.0.0.0"
+PUBLIC_PORT = 18443
+INGRESS_HEALTH = "http://127.0.0.1:18945/health"
+DROPIN_NAME = "zzzzzzzzzz-owner-approved-public.conf"
+
+
+class ActivationError(RuntimeError):
+    pass
+
+
+def canonical(value: object) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+
+
+def sha256_bytes(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def sha256_file(path: Path) -> str:
+    return sha256_bytes(path.read_bytes())
+
+
+def parse_time(value: object, *, field: str) -> datetime:
+    if not isinstance(value, str) or not value.strip():
+        raise ActivationError("APPROVAL_" + field.upper() + "_INVALID")
+    text = value.strip()
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError as exc:
+        raise ActivationError("APPROVAL_" + field.upper() + "_INVALID") from exc
+    if parsed.tzinfo is None:
+        raise ActivationError("APPROVAL_" + field.upper() + "_INVALID")
+    return parsed.astimezone(timezone.utc)
+
+
+def validate_domain(value: object) -> str:
+    if not isinstance(value, str):
+        raise ActivationError("APPROVAL_DOMAIN_INVALID")
+    domain = value.strip().lower()
+    if DOMAIN_RE.fullmatch(domain) is None:
+        raise ActivationError("APPROVAL_DOMAIN_INVALID")
+    return domain
+
+
+def validate_root_file(
+    path: Path,
+    *,
+    exact_mode: int | None = None,
+    max_size: int = 256 * 1024,
+) -> Path:
+    if path.is_symlink() or not path.is_file():
+        raise ActivationError("OWNER_FILE_INVALID")
+    info = path.stat()
+    if info.st_uid != 0 or info.st_gid != 0:
+        raise ActivationError("OWNER_FILE_OWNERSHIP_INVALID")
+    mode = stat.S_IMODE(info.st_mode)
+    if exact_mode is not None and mode != exact_mode:
+        raise ActivationError("OWNER_FILE_MODE_INVALID")
+    if info.st_size < 16 or info.st_size > max_size:
+        raise ActivationError("OWNER_FILE_SIZE_INVALID")
+    return path
+
+
+def certificate_dns_names(cert_path: Path) -> tuple[str, ...]:
+    try:
+        decoded = ssl._ssl._test_decode_cert(str(cert_path))
+    except Exception as exc:
+        raise ActivationError("TLS_CERT_DECODE_INVALID") from exc
+    names = [
+        value
+        for kind, value in decoded.get("subjectAltName", ())
+        if kind == "DNS"
+    ]
+    if not names:
+        for rdn in decoded.get("subject", ()):
+            for key, value in rdn:
+                if key == "commonName":
+                    names.append(value)
+    return tuple(names)
+
+
+def cert_matches_domain(cert_path: Path, domain: str) -> bool:
+    names = certificate_dns_names(cert_path)
+    try:
+        return bool(names) and any(ssl._dnsname_match(name, domain) for name in names)
+    except ssl.CertificateError:
+        return False
+
+
+def validate_tls_pair(
+    cert_path: Path,
+    key_path: Path,
+    domain: str,
+    *,
+    now_fn=time.time,
+) -> dict[str, Any]:
+    validate_root_file(cert_path)
+    validate_root_file(key_path, exact_mode=0o600)
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    context.minimum_version = ssl.TLSVersion.TLSv1_2
+    context.options |= ssl.OP_NO_COMPRESSION
+    try:
+        context.load_cert_chain(str(cert_path), str(key_path))
+    except Exception as exc:
+        raise ActivationError("TLS_CERT_KEY_PAIR_INVALID") from exc
+    if not cert_matches_domain(cert_path, domain):
+        raise ActivationError("TLS_CERT_DOMAIN_MISMATCH")
+    decoded = ssl._ssl._test_decode_cert(str(cert_path))
+    not_after = decoded.get("notAfter")
+    if not not_after:
+        raise ActivationError("TLS_CERT_EXPIRY_MISSING")
+    remaining = ssl.cert_time_to_seconds(not_after) - now_fn()
+    if remaining < 7 * 86400:
+        raise ActivationError("TLS_CERT_EXPIRY_TOO_CLOSE")
+    return {
+        "cert_sha256": sha256_file(cert_path),
+        "remaining_days": int(remaining // 86400),
+    }
+
+
+@dataclass(frozen=True)
+class Approval:
+    approval_id: str
+    candidate_oid: str
+    domain: str
+    bind_host: str
+    port: int
+    tls_cert_sha256: str
+    owner: str
+    issued_at: datetime
+    not_before: datetime
+    expires_at: datetime
+    raw_sha256: str
+
+    @classmethod
+    def load(
+        cls,
+        path: Path,
+        *,
+        candidate_oid: str,
+        cert_path: Path,
+        now: datetime | None = None,
+    ) -> "Approval":
+        validate_root_file(path, exact_mode=0o600, max_size=64 * 1024)
+        try:
+            value = json.loads(path.read_text(encoding="utf-8"))
+        except Exception as exc:
+            raise ActivationError("APPROVAL_JSON_INVALID") from exc
+        if not isinstance(value, Mapping):
+            raise ActivationError("APPROVAL_JSON_INVALID")
+        allowed = {
+            "schema",
+            "approval_id",
+            "candidate_oid",
+            "domain",
+            "public_bind",
+            "bind_host",
+            "port",
+            "tls_cert_sha256",
+            "issued_at",
+            "not_before",
+            "expires_at",
+            "owner",
+        }
+        if set(value) != allowed or value.get("schema") != APPROVAL_SCHEMA:
+            raise ActivationError("APPROVAL_FIELDS_INVALID")
+
+        approval_id = value.get("approval_id")
+        if not isinstance(approval_id, str) or APPROVAL_ID_RE.fullmatch(approval_id) is None:
+            raise ActivationError("APPROVAL_ID_INVALID")
+        if not OID_RE.fullmatch(str(value.get("candidate_oid") or "")):
+            raise ActivationError("APPROVAL_CANDIDATE_INVALID")
+        if value["candidate_oid"] != candidate_oid:
+            raise ActivationError("APPROVAL_CANDIDATE_MISMATCH")
+        if value.get("public_bind") is not True:
+            raise ActivationError("APPROVAL_PUBLIC_BIND_REQUIRED")
+        if value.get("bind_host") != PUBLIC_HOST or value.get("port") != PUBLIC_PORT:
+            raise ActivationError("APPROVAL_BINDING_INVALID")
+
+        cert_sha = str(value.get("tls_cert_sha256") or "")
+        if SHA_RE.fullmatch(cert_sha) is None:
+            raise ActivationError("APPROVAL_CERT_SHA_INVALID")
+        if not cert_path.is_file() or sha256_file(cert_path) != cert_sha:
+            raise ActivationError("APPROVAL_CERT_SHA_MISMATCH")
+
+        domain = validate_domain(value.get("domain"))
+        owner = value.get("owner")
+        if not isinstance(owner, str) or not owner.strip() or len(owner) > 255:
+            raise ActivationError("APPROVAL_OWNER_INVALID")
+
+        issued_at = parse_time(value.get("issued_at"), field="issued_at")
+        not_before = parse_time(value.get("not_before"), field="not_before")
+        expires_at = parse_time(value.get("expires_at"), field="expires_at")
+        current = now or datetime.now(timezone.utc)
+        if not (issued_at <= not_before <= expires_at):
+            raise ActivationError("APPROVAL_TIME_ORDER_INVALID")
+        if current < not_before:
+            raise ActivationError("APPROVAL_NOT_YET_VALID")
+        if current >= expires_at:
+            raise ActivationError("APPROVAL_EXPIRED")
+        if expires_at - issued_at > __import__("datetime").timedelta(days=7):
+            raise ActivationError("APPROVAL_VALIDITY_TOO_LONG")
+
+        return cls(
+            approval_id=approval_id,
+            candidate_oid=candidate_oid,
+            domain=domain,
+            bind_host=PUBLIC_HOST,
+            port=PUBLIC_PORT,
+            tls_cert_sha256=cert_sha,
+            owner=owner.strip(),
+            issued_at=issued_at,
+            not_before=not_before,
+            expires_at=expires_at,
+            raw_sha256=sha256_file(path),
+        )
+
+
+def validate_release(release_root: Path) -> dict[str, Path]:
+    if release_root.is_symlink() or not release_root.is_dir():
+        raise ActivationError("IMMUTABLE_RELEASE_INVALID")
+    edge = release_root / "active-runtime/app/aios_gpt_https_edge.py"
+    controller = release_root / "active-runtime/app/aios_gpt_https_edge_activation.py"
+    for path in (edge, controller):
+        if path.is_symlink() or not path.is_file():
+            raise ActivationError("IMMUTABLE_RELEASE_INVALID")
+    return {"edge": edge, "controller": controller}
+
+
+def render_dropin(approval: Approval, release_root: Path) -> str:
+    app = release_root / "active-runtime/app/aios_gpt_https_edge.py"
+    return (
+        "[Service]\n"
+        f"Environment=AIOS_GPT_EDGE_HOST={approval.bind_host}\n"
+        f"Environment=AIOS_GPT_EDGE_PORT={approval.port}\n"
+        "Environment=AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1\n"
+        f"Environment=AIOS_GPT_EDGE_PUBLIC_DOMAIN={approval.domain}\n"
+        "ExecStart=\n"
+        f"ExecStart=/usr/bin/python3 -B {app}\n"
+        f"ReadOnlyPaths={release_root}\n"
+    )
+
+
+def atomic_write(path: Path, data: str, mode: int = 0o600) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, name = tempfile.mkstemp(prefix="." + path.name + ".", dir=path.parent)
+    try:
+        os.fchmod(fd, mode)
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(data)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(name, path)
+    finally:
+        if os.path.exists(name):
+            os.unlink(name)
+
+
+class Systemd:
+    def run(self, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
+        return subprocess.run(
+            ["/usr/bin/systemctl", *args],
+            check=check,
+            text=True,
+            capture_output=True,
+            timeout=30,
+        )
+
+    def active(self, unit: str) -> bool:
+        return self.run("is-active", "--quiet", unit, check=False).returncode == 0
+
+    def enabled(self, unit: str) -> bool:
+        return self.run("is-enabled", "--quiet", unit, check=False).returncode == 0
+
+    def show(self, unit: str, prop: str) -> str:
+        return self.run("show", unit, f"--property={prop}", "--value").stdout.strip()
+
+
+def require_ingress_ready(systemd: Systemd) -> None:
+    if not systemd.active(INGRESS_UNIT):
+        raise ActivationError("INGRESS_SERVICE_NOT_ACTIVE")
+    try:
+        with urllib.request.urlopen(INGRESS_HEALTH, timeout=5) as response:
+            if response.status != 200:
+                raise ActivationError("INGRESS_HEALTH_FAILED")
+            value = json.loads(response.read().decode())
+    except ActivationError:
+        raise
+    except Exception as exc:
+        raise ActivationError("INGRESS_HEALTH_FAILED") from exc
+    if value.get("status") != "ok":
+        raise ActivationError("INGRESS_HEALTH_FAILED")
+
+
+def tls_probe(domain: str, cert_path: Path, *, port: int = PUBLIC_PORT) -> None:
+    context = ssl.create_default_context(cafile=str(cert_path))
+    context.check_hostname = True
+    context.verify_mode = ssl.CERT_REQUIRED
+    try:
+        with socket.create_connection(("127.0.0.1", port), timeout=5) as sock:
+            with context.wrap_socket(sock, server_hostname=domain) as tls_sock:
+                if tls_sock.version() not in {"TLSv1.2", "TLSv1.3"}:
+                    raise ActivationError("EDGE_TLS_VERSION_INVALID")
+    except ActivationError:
+        raise
+    except Exception as exc:
+        raise ActivationError("EDGE_TLS_PROBE_FAILED") from exc
+
+
+def receipt_for(
+    approval: Approval,
+    release_root: Path,
+    *,
+    state: str,
+) -> dict[str, Any]:
+    return {
+        "schema": STATE_SCHEMA,
+        "state": state,
+        "approval_id": approval.approval_id,
+        "approval_sha256": approval.raw_sha256,
+        "candidate_oid": approval.candidate_oid,
+        "domain": approval.domain,
+        "bind_host": approval.bind_host,
+        "port": approval.port,
+        "tls_cert_sha256": approval.tls_cert_sha256,
+        "release_root": str(release_root),
+        "updated_at": datetime.now(timezone.utc).isoformat(),
+    }
+
+
+def activate(
+    *,
+    approval: Approval,
+    release_root: Path,
+    cert_path: Path,
+    key_path: Path,
+    dropin_path: Path,
+    state_path: Path,
+    systemd: Systemd,
+) -> dict[str, Any]:
+    if os.geteuid() != 0:
+        raise ActivationError("ROOT_REQUIRED")
+    release = validate_release(release_root)
+    app = release["edge"]
+    require_ingress_ready(systemd)
+    validate_tls_pair(cert_path, key_path, approval.domain)
+
+    if state_path.is_file():
+        try:
+            existing = json.loads(state_path.read_text(encoding="utf-8"))
+        except Exception:
+            existing = {}
+        if (
+            existing.get("schema") == STATE_SCHEMA
+            and existing.get("state") == "ACTIVE"
+            and existing.get("approval_sha256") == approval.raw_sha256
+            and existing.get("candidate_oid") == approval.candidate_oid
+            and systemd.active(EDGE_UNIT)
+        ):
+            return existing
+
+    dropin = render_dropin(approval, release_root)
+    modified = False
+    try:
+        atomic_write(dropin_path, dropin, mode=0o644)
+        modified = True
+        systemd.run("daemon-reload")
+        systemd.run("enable", EDGE_UNIT)
+        systemd.run("restart", EDGE_UNIT)
+        if not systemd.active(EDGE_UNIT):
+            raise ActivationError("EDGE_SERVICE_NOT_ACTIVE")
+        exec_start = systemd.show(EDGE_UNIT, "ExecStart")
+        environment = systemd.show(EDGE_UNIT, "Environment")
+        if str(app) not in exec_start:
+            raise ActivationError("EDGE_EFFECTIVE_EXECSTART_INVALID")
+        for expected in (
+            f"AIOS_GPT_EDGE_HOST={approval.bind_host}",
+            f"AIOS_GPT_EDGE_PORT={approval.port}",
+            "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1",
+            f"AIOS_GPT_EDGE_PUBLIC_DOMAIN={approval.domain}",
+        ):
+            if expected not in environment:
+                raise ActivationError("EDGE_EFFECTIVE_ENVIRONMENT_INVALID")
+        tls_probe(approval.domain, cert_path, port=approval.port)
+        receipt = receipt_for(approval, release_root, state="ACTIVE")
+        atomic_write(state_path, json.dumps(receipt, indent=2, sort_keys=True) + "\n")
+        return receipt
+    except Exception:
+        if modified:
+            rollback(
+                dropin_path=dropin_path,
+                state_path=state_path,
+                systemd=systemd,
+                reason="activation_failed",
+            )
+        raise
+
+
+def rollback(
+    *,
+    dropin_path: Path,
+    state_path: Path,
+    systemd: Systemd,
+    reason: str,
+) -> dict[str, Any]:
+    if os.geteuid() != 0:
+        raise ActivationError("ROOT_REQUIRED")
+    systemd.run("stop", EDGE_UNIT, check=False)
+    systemd.run("disable", EDGE_UNIT, check=False)
+    dropin_path.unlink(missing_ok=True)
+    systemd.run("daemon-reload")
+    if systemd.active(EDGE_UNIT):
+        raise ActivationError("EDGE_ROLLBACK_SERVICE_STILL_ACTIVE")
+    previous: dict[str, Any] = {}
+    if state_path.is_file():
+        try:
+            previous = json.loads(state_path.read_text(encoding="utf-8"))
+        except Exception:
+            previous = {}
+    receipt = {
+        "schema": STATE_SCHEMA,
+        "state": "ROLLED_BACK",
+        "reason": reason[:255],
+        "candidate_oid": previous.get("candidate_oid"),
+        "approval_id": previous.get("approval_id"),
+        "domain": previous.get("domain"),
+        "updated_at": datetime.now(timezone.utc).isoformat(),
+    }
+    atomic_write(state_path, json.dumps(receipt, indent=2, sort_keys=True) + "\n")
+    return receipt
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("command", choices=("check", "activate", "rollback"))
+    parser.add_argument("--candidate")
+    parser.add_argument("--approval", default="/etc/aios-gitops/gpt-https-edge.approved")
+    parser.add_argument("--tls-cert", default="/etc/aios-gitops/gpt-https-edge-tls.crt")
+    parser.add_argument("--tls-key", default="/etc/aios-gitops/gpt-https-edge-tls.key")
+    parser.add_argument("--release-root")
+    parser.add_argument(
+        "--dropin",
+        default=f"/etc/systemd/system/{EDGE_UNIT}.d/{DROPIN_NAME}",
+    )
+    parser.add_argument(
+        "--state",
+        default="/var/lib/aios-gpt-https-edge-activation/state.json",
+    )
+    parser.add_argument("--reason", default="owner_requested")
+    args = parser.parse_args()
+
+    systemd = Systemd()
+    if args.command == "rollback":
+        value = rollback(
+            dropin_path=Path(args.dropin),
+            state_path=Path(args.state),
+            systemd=systemd,
+            reason=args.reason,
+        )
+        print(json.dumps(value, indent=2, sort_keys=True))
+        return 0
+
+    candidate = str(args.candidate or "")
+    if OID_RE.fullmatch(candidate) is None:
+        raise SystemExit("CANDIDATE_REQUIRED")
+    release_root = (
+        Path(args.release_root)
+        if args.release_root
+        else Path(f"/opt/aios/releases/{candidate}-gpt-https-edge-public")
+    )
+    cert_path = Path(args.tls_cert)
+    key_path = Path(args.tls_key)
+    approval = Approval.load(
+        Path(args.approval),
+        candidate_oid=candidate,
+        cert_path=cert_path,
+    )
+    tls = validate_tls_pair(cert_path, key_path, approval.domain)
+    release = validate_release(release_root)
+    require_ingress_ready(systemd)
+
+    if args.command == "check":
+        value = {
+            "status": "READY",
+            "candidate_oid": candidate,
+            "approval_id": approval.approval_id,
+            "approval_sha256": approval.raw_sha256,
+            "domain": approval.domain,
+            "bind_host": approval.bind_host,
+            "port": approval.port,
+            "edge_sha256": sha256_file(release["edge"]),
+            "controller_sha256": sha256_file(release["controller"]),
+            **tls,
+            "public_network_mutation": False,
+        }
+        print(json.dumps(value, indent=2, sort_keys=True))
+        return 0
+
+    value = activate(
+        approval=approval,
+        release_root=release_root,
+        cert_path=cert_path,
+        key_path=key_path,
+        dropin_path=Path(args.dropin),
+        state_path=Path(args.state),
+        systemd=systemd,
+    )
+    print(json.dumps(value, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,114 @@
+# Owner-approved GPT HTTPS edge activation and rollback
+
+## Scope
+
+This controller is the final host-side control mechanism before the disabled-by-default GPT HTTPS
+edge may bind publicly.
+
+Repository implementation and public activation remain separate states:
+
+```text
+IMPLEMENTED
+≠
+OWNER_APPROVED
+≠
+ACTIVATED
+≠
+RUNTIME_VERIFIED
+```
+
+The controller never creates owner approval, TLS assets, DNS, firewall rules or ingress bearer
+credentials.
+
+## Approval binding
+
+The owner approval file is root-owned mode `0600` and has schema:
+
+```json
+{
+  "schema": "aios.gpt-https-edge.public-approval.v1",
+  "approval_id": "owner-public-edge-...",
+  "candidate_oid": "<exact governed edge candidate>",
+  "domain": "tasks.example.com",
+  "public_bind": true,
+  "bind_host": "0.0.0.0",
+  "port": 18443,
+  "tls_cert_sha256": "<exact certificate sha256>",
+  "issued_at": "2026-08-19T00:00:00Z",
+  "not_before": "2026-08-19T00:00:00Z",
+  "expires_at": "2026-08-20T00:00:00Z",
+  "owner": "<owner identity>"
+}
+```
+
+Approval lifetime is at most seven days. Candidate/domain/bind/port/certificate are immutable within
+one approval.
+
+Changing any of those requires a new owner approval.
+
+## Check
+
+```text
+aios_gpt_https_edge_activation.py check --candidate <GE>
+```
+
+Checks:
+
+- approval ownership/mode/schema/time window;
+- exact candidate binding;
+- exact certificate hash;
+- certificate/private-key pair;
+- certificate hostname;
+- certificate remaining lifetime;
+- loopback GPT ingress service health.
+
+It performs no public network mutation.
+
+## Activate
+
+`activate` additionally:
+
+- writes one exact systemd public-bind drop-in atomically;
+- enables/restarts only the HTTPS edge unit;
+- verifies effective ExecStart points at the immutable candidate release;
+- verifies effective public-bind/domain environment;
+- performs a TLS handshake to `127.0.0.1:18443` using the approved public hostname;
+- writes a durable activation receipt.
+
+Any failure after drop-in mutation automatically invokes rollback.
+
+Exact replay of an already-active matching approval returns the durable receipt without restarting
+or creating duplicate state.
+
+## Rollback
+
+`rollback`:
+
+- stops and disables only the HTTPS edge;
+- removes only the controller-owned public-bind drop-in;
+- daemon-reloads systemd;
+- verifies edge inactive;
+- writes a durable rollback receipt.
+
+It does not delete:
+
+- owner approval;
+- TLS certificate/private key;
+- ingress bearer token;
+- internal loopback ingress;
+- Task/Job state;
+- Git branches.
+
+Rollback is therefore replay-safe and does not require reconstructing secrets.
+
+## Not covered
+
+This controller intentionally does not own:
+
+- DNS records;
+- host firewall/NAT/security-group changes;
+- certificate issuance/renewal;
+- OpenAI Custom GPT domain verification;
+- public Internet acceptance from outside the host.
+
+Those remain separate owner/environment gates.
@@ -0,0 +1,362 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+import hashlib
+import json
+from pathlib import Path
+import sys
+import subprocess
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+import aios_gpt_https_edge_activation as activation
+
+
+CANDIDATE="69a51d50c7f6cd9045346d41e338acebe4b6f1e8"
+NOW=datetime(2026,8,19,2,0,0,tzinfo=timezone.utc)
+
+
+def approval_value(cert_sha: str, **overrides):
+    value={
+        "schema":"aios.gpt-https-edge.public-approval.v1",
+        "approval_id":"owner-edge-approval-0001",
+        "candidate_oid":CANDIDATE,
+        "domain":"tasks.example.com",
+        "public_bind":True,
+        "bind_host":"0.0.0.0",
+        "port":18443,
+        "tls_cert_sha256":cert_sha,
+        "issued_at":"2026-08-19T01:00:00Z",
+        "not_before":"2026-08-19T01:00:00Z",
+        "expires_at":"2026-08-20T01:00:00Z",
+        "owner":"owner@example",
+    }
+    value.update(overrides)
+    return value
+
+
+def write_owner_file(path: Path, data: bytes, mode=0o600):
+    path.write_bytes(data)
+    path.chmod(mode)
+    return path
+
+
+def load_approval(tmp_path, monkeypatch, **overrides):
+    tmp_path.mkdir(parents=True,exist_ok=True)
+    cert=write_owner_file(tmp_path/"cert.pem",b"certificate-material-"*4,0o600)
+    value=approval_value(hashlib.sha256(cert.read_bytes()).hexdigest(),**overrides)
+    approval=write_owner_file(
+        tmp_path/"approval.json",
+        (json.dumps(value,sort_keys=True)+"\n").encode(),
+        0o600,
+    )
+    monkeypatch.setattr(
+        activation,
+        "validate_root_file",
+        lambda path, **kwargs: path,
+    )
+    return activation.Approval.load(
+        approval,
+        candidate_oid=CANDIDATE,
+        cert_path=cert,
+        now=NOW,
+    ),approval,cert
+
+
+class FakeSystemd:
+    def __init__(self):
+        self.active_units={activation.INGRESS_UNIT}
+        self.enabled_units=set()
+        self.calls=[]
+        self.exec_start=""
+        self.environment=""
+        self.fail_restart=False
+
+    def run(self,*args,check=True):
+        self.calls.append(args)
+        class Result:
+            returncode=0
+            stdout=""
+        if args[:2]==("restart",activation.EDGE_UNIT):
+            if self.fail_restart:
+                raise subprocess.CalledProcessError(1,args)
+            self.active_units.add(activation.EDGE_UNIT)
+        elif args[:2]==("stop",activation.EDGE_UNIT):
+            self.active_units.discard(activation.EDGE_UNIT)
+        elif args[:2]==("enable",activation.EDGE_UNIT):
+            self.enabled_units.add(activation.EDGE_UNIT)
+        elif args[:2]==("disable",activation.EDGE_UNIT):
+            self.enabled_units.discard(activation.EDGE_UNIT)
+        return Result()
+
+    def active(self,unit):
+        return unit in self.active_units
+
+    def enabled(self,unit):
+        return unit in self.enabled_units
+
+    def show(self,unit,prop):
+        if prop=="ExecStart":
+            return self.exec_start
+        if prop=="Environment":
+            return self.environment
+        return ""
+
+
+def test_approval_binds_candidate_domain_bind_port_and_cert(tmp_path,monkeypatch):
+    approval,_,cert=load_approval(tmp_path,monkeypatch)
+    assert approval.candidate_oid==CANDIDATE
+    assert approval.domain=="tasks.example.com"
+    assert approval.bind_host=="0.0.0.0"
+    assert approval.port==18443
+    assert approval.tls_cert_sha256==hashlib.sha256(cert.read_bytes()).hexdigest()
+    assert approval.expires_at>approval.not_before
+
+
+def test_approval_rejects_candidate_mismatch(tmp_path,monkeypatch):
+    cert=write_owner_file(tmp_path/"cert.pem",b"certificate-material-"*4)
+    value=approval_value(hashlib.sha256(cert.read_bytes()).hexdigest())
+    approval=write_owner_file(tmp_path/"approval.json",json.dumps(value).encode())
+    monkeypatch.setattr(activation,"validate_root_file",lambda path,**kwargs:path)
+    with pytest.raises(activation.ActivationError,match="CANDIDATE_MISMATCH"):
+        activation.Approval.load(
+            approval,
+            candidate_oid="0"*40,
+            cert_path=cert,
+            now=NOW,
+        )
+
+
+def test_approval_rejects_cert_hash_mismatch(tmp_path,monkeypatch):
+    cert=write_owner_file(tmp_path/"cert.pem",b"certificate-material-"*4)
+    value=approval_value("a"*64)
+    approval=write_owner_file(tmp_path/"approval.json",json.dumps(value).encode())
+    monkeypatch.setattr(activation,"validate_root_file",lambda path,**kwargs:path)
+    with pytest.raises(activation.ActivationError,match="CERT_SHA_MISMATCH"):
+        activation.Approval.load(
+            approval,
+            candidate_oid=CANDIDATE,
+            cert_path=cert,
+            now=NOW,
+        )
+
+
+def test_approval_rejects_expired_not_yet_valid_and_long_lived(tmp_path,monkeypatch):
+    with pytest.raises(activation.ActivationError,match="APPROVAL_EXPIRED"):
+        load_approval(
+            tmp_path/"expired",
+            monkeypatch,
+            issued_at="2026-08-17T01:00:00Z",
+            not_before="2026-08-17T01:00:00Z",
+            expires_at="2026-08-18T01:00:00Z",
+        )
+    # Use distinct temp roots because helper writes fixed names.
+
+    with pytest.raises(activation.ActivationError,match="NOT_YET_VALID"):
+        load_approval(
+            tmp_path/"future",
+            monkeypatch,
+            issued_at="2026-08-19T03:00:00Z",
+            not_before="2026-08-19T03:00:00Z",
+            expires_at="2026-08-20T03:00:00Z",
+        )
+    with pytest.raises(activation.ActivationError,match="VALIDITY_TOO_LONG"):
+        load_approval(
+            tmp_path/"long",
+            monkeypatch,
+            issued_at="2026-08-18T01:00:00Z",
+            not_before="2026-08-18T01:00:00Z",
+            expires_at="2026-08-26T01:00:00Z",
+        )
+
+
+
+def test_validate_release_requires_edge_and_controller_and_denies_symlink(tmp_path):
+    release=tmp_path/"release"
+    app=release/"active-runtime/app"
+    app.mkdir(parents=True)
+    edge_file=app/"aios_gpt_https_edge.py"
+    controller=app/"aios_gpt_https_edge_activation.py"
+    edge_file.write_text("edge\n")
+    controller.write_text("controller\n")
+    value=activation.validate_release(release)
+    assert value["edge"]==edge_file
+    assert value["controller"]==controller
+    controller.unlink()
+    with pytest.raises(activation.ActivationError,match="IMMUTABLE_RELEASE_INVALID"):
+        activation.validate_release(release)
+
+def test_render_dropin_is_exact_and_contains_no_secret(tmp_path,monkeypatch):
+    approval,_,_=load_approval(tmp_path,monkeypatch)
+    release=Path("/opt/aios/releases")/(CANDIDATE+"-gpt-https-edge-public")
+    text=activation.render_dropin(approval,release)
+    assert "AIOS_GPT_EDGE_HOST=0.0.0.0" in text
+    assert "AIOS_GPT_EDGE_PORT=18443" in text
+    assert "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1" in text
+    assert "AIOS_GPT_EDGE_PUBLIC_DOMAIN=tasks.example.com" in text
+    assert CANDIDATE in text
+    for forbidden in ("Bearer ","agent_task_control","tls_key","gpt_ingress_token"):
+        assert forbidden not in text
+
+
+def test_receipt_contains_binding_but_no_owner_secret_material(tmp_path,monkeypatch):
+    approval,_,_=load_approval(tmp_path,monkeypatch)
+    receipt=activation.receipt_for(
+        approval,
+        Path("/opt/aios/releases/test"),
+        state="ACTIVE",
+    )
+    assert receipt["approval_id"]==approval.approval_id
+    assert receipt["candidate_oid"]==CANDIDATE
+    assert receipt["domain"]=="tasks.example.com"
+    serialized=json.dumps(receipt)
+    assert "Bearer " not in serialized
+    assert "private" not in serialized.lower()
+
+
+def test_activate_is_idempotent_for_same_active_approval(tmp_path,monkeypatch):
+    approval,_,cert=load_approval(tmp_path,monkeypatch)
+    release=tmp_path/"release"
+    app=release/"active-runtime/app/aios_gpt_https_edge.py"
+    controller=release/"active-runtime/app/aios_gpt_https_edge_activation.py"
+    app.parent.mkdir(parents=True)
+    app.write_text("print('x')\n")
+    controller.write_text("print('controller')\n")
+    key=write_owner_file(tmp_path/"key.pem",b"private-key-material-"*4,0o600)
+    dropin=tmp_path/"edge.d"/activation.DROPIN_NAME
+    state=tmp_path/"state.json"
+    systemd=FakeSystemd()
+
+    monkeypatch.setattr(activation.os,"geteuid",lambda:0)
+    monkeypatch.setattr(activation,"require_ingress_ready",lambda _:None)
+    monkeypatch.setattr(activation,"validate_tls_pair",lambda *a,**k:{"cert_sha256":approval.tls_cert_sha256})
+    monkeypatch.setattr(activation,"tls_probe",lambda *a,**k:None)
+
+    systemd.exec_start=str(app)
+    systemd.environment=(
+        "AIOS_GPT_EDGE_HOST=0.0.0.0 "
+        "AIOS_GPT_EDGE_PORT=18443 "
+        "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1 "
+        "AIOS_GPT_EDGE_PUBLIC_DOMAIN=tasks.example.com"
+    )
+    first=activation.activate(
+        approval=approval,
+        release_root=release,
+        cert_path=cert,
+        key_path=key,
+        dropin_path=dropin,
+        state_path=state,
+        systemd=systemd,
+    )
+    calls_after_first=len(systemd.calls)
+    second=activation.activate(
+        approval=approval,
+        release_root=release,
+        cert_path=cert,
+        key_path=key,
+        dropin_path=dropin,
+        state_path=state,
+        systemd=systemd,
+    )
+    assert first["state"]=="ACTIVE"
+    assert second==json.loads(state.read_text())
+    assert len(systemd.calls)==calls_after_first
+
+
+def test_activate_failure_after_mutation_rolls_back(tmp_path,monkeypatch):
+    approval,_,cert=load_approval(tmp_path,monkeypatch)
+    release=tmp_path/"release"
+    app=release/"active-runtime/app/aios_gpt_https_edge.py"
+    controller=release/"active-runtime/app/aios_gpt_https_edge_activation.py"
+    app.parent.mkdir(parents=True)
+    app.write_text("print('x')\n")
+    controller.write_text("print('controller')\n")
+    key=write_owner_file(tmp_path/"key.pem",b"private-key-material-"*4,0o600)
+    dropin=tmp_path/"edge.d"/activation.DROPIN_NAME
+    state=tmp_path/"state.json"
+    systemd=FakeSystemd()
+    systemd.fail_restart=True
+
+    monkeypatch.setattr(activation.os,"geteuid",lambda:0)
+    monkeypatch.setattr(activation,"require_ingress_ready",lambda _:None)
+    monkeypatch.setattr(activation,"validate_tls_pair",lambda *a,**k:{"cert_sha256":approval.tls_cert_sha256})
+    monkeypatch.setattr(activation,"tls_probe",lambda *a,**k:None)
+
+    with pytest.raises(subprocess.CalledProcessError):
+        activation.activate(
+            approval=approval,
+            release_root=release,
+            cert_path=cert,
+            key_path=key,
+            dropin_path=dropin,
+            state_path=state,
+            systemd=systemd,
+        )
+    assert not dropin.exists()
+    assert not systemd.active(activation.EDGE_UNIT)
+    assert json.loads(state.read_text())["state"]=="ROLLED_BACK"
+
+
+def test_rollback_is_replay_safe_and_leaves_owner_assets_alone(tmp_path,monkeypatch):
+    approval_file=tmp_path/"approval.json"
+    cert=tmp_path/"cert.pem"
+    key=tmp_path/"key.pem"
+    for path in (approval_file,cert,key):
+        path.write_text("owner-asset"*4)
+    dropin=tmp_path/"edge.d"/activation.DROPIN_NAME
+    dropin.parent.mkdir(parents=True)
+    dropin.write_text("[Service]\n")
+    state=tmp_path/"state.json"
+    state.write_text(json.dumps({
+        "schema":activation.STATE_SCHEMA,
+        "state":"ACTIVE",
+        "candidate_oid":CANDIDATE,
+        "approval_id":"owner-edge-approval-0001",
+        "domain":"tasks.example.com",
+    }))
+    systemd=FakeSystemd()
+    systemd.active_units.add(activation.EDGE_UNIT)
+    systemd.enabled_units.add(activation.EDGE_UNIT)
+    monkeypatch.setattr(activation.os,"geteuid",lambda:0)
+
+    first=activation.rollback(
+        dropin_path=dropin,
+        state_path=state,
+        systemd=systemd,
+        reason="owner_requested",
+    )
+    second=activation.rollback(
+        dropin_path=dropin,
+        state_path=state,
+        systemd=systemd,
+        reason="owner_requested",
+    )
+    assert first["state"]=="ROLLED_BACK"
+    assert second["state"]=="ROLLED_BACK"
+    assert not dropin.exists()
+    assert not systemd.active(activation.EDGE_UNIT)
+    assert not systemd.enabled(activation.EDGE_UNIT)
+    for path in (approval_file,cert,key):
+        assert path.exists()
+
+
+def test_activation_source_never_creates_owner_approval_or_tls_secret():
+    source=(ROOT/"active-runtime/app/aios_gpt_https_edge_activation.py").read_text()
+    assert "secrets.token" not in source
+    assert "openssl" not in source.lower()
+    assert "BEGIN PRIVATE KEY" not in source
+    assert "gpt-https-edge.approved" in source
+    assert "gpt-https-edge-tls.key" in source
+
+
+def test_docs_separate_implementation_activation_and_runtime_verification():
+    doc=(ROOT/"docs/superpowers/specs/2026-08-19-gpt-https-edge-activation.md").read_text()
+    assert "IMPLEMENTED" in doc
+    assert "OWNER_APPROVED" in doc
+    assert "ACTIVATED" in doc
+    assert "RUNTIME_VERIFIED" in doc
+    assert "does not own" in doc.lower()