363 lines
13 KiB
Python
363 lines
13 KiB
Python
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()
|