feat(frontdoor): git-push deploy flow + grant-bound remote terminal + login/session

Makefile

@@ -45,6 +45,12 @@ lease: ## pull/lease scheduler demo: workers pull WUs, crash-stop re-lending, or
 sphere: ## data-sphere demo: immutable dm-verity sphere, construction-tenancy, intent x link x durability
 	cd tools && python3 data_sphere.py
 
+push: ## git-push deploy flow demo: build -> deploy -> per-branch preview
+	cd tools && python3 deploy_flow.py
+
+login: ## login/session demo: authenticate the front door (fail-closed)
+	cd tools && python3 login.py
+
 provision: ## provisioning demo: tenant broker + entitlement tier + BSS meter + /me
 	cd tools && python3 provisioning.py
 

capd/git-push-deploy.mesh.capd.json

@@ -6,6 +6,9 @@
   "description": "Vercel and Heroku are the same move: detect the app from source, build it into a runnable image without a Dockerfile, deploy it, and give a preview environment per branch. This delivers that ergonomic sovereign and open, via Cloud Native Buildpacks / Paketo (`pack build`): source -> detect -> reproducible SBOM'd OCI image (a data sphere, SLSA-attestable) -> a workload the executor dispatches into a DevSpace -> a Signadot-style sandbox as the per-branch preview -> the fail-closed promotion gate as dev->prod. No Docker daemon, no hand-written Dockerfile; the build is reproducible + attestable, the runtime is our mesh (including sovereign GPU inference), and the data stays in residency-fenced data spheres.",
   "links": {
     "engine": "tools/buildpack.py",
+    "flow": "tools/deploy_flow.py",
+    "login": "tools/login.py",
+    "remote_terminal": "tools/devmode.py (attach_command — grant-bound PTY)",
     "executor": "tools/executor.py",
     "preview": "tools/devspace.py",
     "promotion": "tools/promotion_gate.py",

tools/deploy_flow.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+"""Git-push deploy flow — build -> deploy -> per-branch preview, one real flow.
+
+The thin ergonomic wrapper that makes the whole stack `git push`-to-deploy. On a push, detect+build
+the source (buildpack, no Dockerfile) into a reproducible OCI image, turn it into a workload the
+compute plane places + the executor dispatches behind a Grant, and open a per-branch PREVIEW — a
+Signadot-style sandbox that shares the baseline (route `x-sandbox-routing-key: <branch>` to the fork).
+Promotion of a preview to prod is the fail-closed promotion gate. That is Vercel/Heroku's
+"push a branch, get a preview URL," sovereign and governed.
+"""
+from __future__ import annotations
+
+import buildpack as bp
+import devspace as dv
+
+
+def on_push(*, tenant: str, user: str, repo: str, branch: str, source_files: list,
+            sensitivity: str = "normal", app: str | None = None) -> dict:
+    """Handle a push: build the source, produce a deployable workload, and open a per-branch preview."""
+    build = bp.build_plan(source_files=source_files, app_name=f"{repo}-{branch}")
+    if not build.get("ok"):
+        return {"status": "build-failed", "reason": build["reason"], "branch": branch}
+
+    workload = bp.deploy_workload(build, kind="service", sensitivity=sensitivity)
+    namespace = "ds-" + dv._slug(tenant, user, app or "default")
+    preview = dv.sandbox_manifests(baseline=repo, image=build["image"], routing_key=branch,
+                                   namespace=namespace)
+    return {
+        "status": "deployed", "branch": branch, "language": build["language"],
+        "image": build["image"], "build_digest": build["image_digest"], "workload": workload,
+        "preview": {"namespace": namespace, "baseline": repo, "routing_key": branch,
+                    "route_header": f"x-sandbox-routing-key: {branch}", "manifests": preview},
+        "promote_via": "fail-closed promotion gate (sealed APPROVE verdict)",
+    }
+
+
+if __name__ == "__main__":
+    import json
+    out = on_push(tenant="acme", user="alice", repo="productpage", branch="pr-42",
+                  source_files=["package.json", "server.js"])
+    print(json.dumps({"status": out["status"], "image": out["image"],
+                      "preview_route": out["preview"]["route_header"],
+                      "preview_fork": out["preview"]["manifests"][0]["metadata"]["name"],
+                      "promote_via": out["promote_via"]}, indent=2))

tools/devmode.py

@@ -50,6 +50,22 @@ def port_forward_command(*, namespace: str, pod: str, ports: list, context: str
     return ["kubectl", *_ctx(context), "port-forward", "-n", namespace, f"pod/{pod}", *maps]
 
 
+def attach_command(*, namespace: str, pod: str, grant: dict, verifier, session_id: str,
+                   container: str = "dev", context: str | None = None) -> dict:
+    """Grant-bound remote terminal (Nocalhost AppA-terminal / cloud-shell attach). The fog-node Policy
+    Gate re-verifies the Grant (effect exec + op pty.attach, session-bound) BEFORE any PTY is opened —
+    fail-closed. Returns {authorized, command|reason, redactions}."""
+    import mcp_a2a_grant as g
+    check = g.verify_grant(grant, session_id=session_id, verifier=verifier,
+                           requested_effect="exec", requested_op="pty.attach")
+    if not check["result"]["valid"]:
+        return {"authorized": False, "reason": check["result"]["reason"]}
+    return {"authorized": True,
+            "command": ["kubectl", *_ctx(context), "exec", "-it", "-n", namespace,
+                        f"pod/{pod}", "-c", container, "--", "/bin/sh"],
+            "redactions": check.get("redactions", [])}
+
+
 def devmode_plan(*, workload: str, namespace: str, local_dir: str, ports: list,
                  dev_image: str = "python:3.12-alpine", run_cmd: str | None = None,
                  context: str | None = None, grant_id: str | None = None) -> dict:

tools/login.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+"""Login / session — authenticate the twin/box front door.
+
+A governed session surface. Real SSO is OIDC/FIDO2 (the production swap — the same IdP the cloud-shell
+fog spec already names); this is the session core: authenticate a user against a credential verifier,
+issue a signed, expiring session bound to the user + tier, and verify it fail-closed. The session is
+the bearer the portal, provisioning, and grant issuance trust as "who is this" — so the phone hitting
+the twin is authenticated, not open.
+"""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+from datetime import datetime, timedelta, timezone
+
+
+def _canon(body: dict) -> bytes:
+    return json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def _sign(key: bytes, body: dict) -> str:
+    return hmac.new(key, _canon(body), hashlib.sha256).hexdigest()
+
+
+def _iso(dt: datetime) -> str:
+    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _parse(s: str) -> datetime:
+    return datetime.fromisoformat(s.replace("Z", "+00:00"))
+
+
+def issue_session(*, user: str, tier: str, key: bytes, ttl_s: float = 3600.0,
+                  now: datetime | None = None) -> dict:
+    now = now or datetime.now(timezone.utc)
+    body = {"user": user, "tier": tier, "issued_at": _iso(now),
+            "expires_at": _iso(now + timedelta(seconds=float(ttl_s))),
+            "session_id": "sess_" + hashlib.sha256(f"{user}:{_iso(now)}".encode()).hexdigest()[:10]}
+    return {**body, "sig": _sign(key, body)}
+
+
+def authenticate(*, user: str, credential: str, credential_check, key: bytes, tier: str = "pro",
+                 ttl_s: float = 3600.0, now: datetime | None = None) -> dict | None:
+    """Fail-closed: `credential_check(user, credential) -> bool`. Bad credentials -> no session.
+    Swap credential_check for an OIDC/FIDO2 verifier in production."""
+    if not credential_check(user, credential):
+        return None
+    return issue_session(user=user, tier=tier, key=key, ttl_s=ttl_s, now=now)
+
+
+def verify_session(session: dict, *, key: bytes, now: datetime | None = None) -> dict:
+    """Fail-closed session verification. Returns {valid, reason?, user?, tier?}."""
+    now = now or datetime.now(timezone.utc)
+    body = {k: v for k, v in session.items() if k != "sig"}
+    if not hmac.compare_digest(_sign(key, body), session.get("sig", "")):
+        return {"valid": False, "reason": "signature invalid — session tampered or wrong key"}
+    if now > _parse(session.get("expires_at", "1970-01-01T00:00:00Z")):
+        return {"valid": False, "reason": "session expired"}
+    return {"valid": True, "user": session["user"], "tier": session["tier"]}
+
+
+if __name__ == "__main__":
+    key = b"login-demo-key"
+    # a trivial credential check for the demo; production swaps in OIDC/FIDO2.
+    creds = {"alice": "s3cret"}
+    ok = authenticate(user="alice", credential="s3cret", key=key, tier="pro",
+                      credential_check=lambda u, c: creds.get(u) == c)
+    bad = authenticate(user="alice", credential="wrong", key=key,
+                       credential_check=lambda u, c: creds.get(u) == c)
+    print(json.dumps({"authenticated": bool(ok), "bad_creds_rejected": bad is None,
+                      "verify": verify_session(ok, key=key)}, indent=2))

tools/test_deploy_flow.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+"""Tests for the git-push deploy flow (build -> deploy -> per-branch preview)."""
+import deploy_flow as df
+
+
+def test_push_builds_deploys_and_opens_a_per_branch_preview():
+    out = df.on_push(tenant="acme", user="alice", repo="productpage", branch="pr-42",
+                     source_files=["package.json", "server.js"])
+    assert out["status"] == "deployed" and out["language"] == "node"
+    assert out["workload"]["image"] == out["image"]
+    # the preview is a Signadot-style sandbox routed by the branch
+    assert out["preview"]["routing_key"] == "pr-42"
+    assert out["preview"]["route_header"] == "x-sandbox-routing-key: pr-42"
+    fork = out["preview"]["manifests"][0]
+    assert fork["kind"] == "Deployment" and "sbx-pr-42" in fork["metadata"]["name"]
+
+
+def test_push_with_unbuildable_source_fails_closed():
+    out = df.on_push(tenant="acme", user="alice", repo="x", branch="main",
+                     source_files=["notes.txt"])
+    assert out["status"] == "build-failed" and "no buildpack matched" in out["reason"]
+
+
+def test_preview_namespace_is_the_tenant_devspace():
+    out = df.on_push(tenant="acme", user="bob", repo="api", branch="feat", source_files=["go.mod"])
+    assert out["preview"]["namespace"] == "ds-acme-bob-default"
+
+
+if __name__ == "__main__":
+    import sys
+    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
+    for fn in fns:
+        fn()
+    print(f"ok: {len(fns)} deploy-flow tests passed")
+    sys.exit(0)

tools/test_devmode.py

@@ -1,6 +1,39 @@
 #!/usr/bin/env python3
-"""Tests for dev-mode manifest/command generation (the pure core of the inner loop)."""
+"""Tests for dev-mode manifest/command generation (the pure core of the inner loop) + the grant-bound
+remote terminal."""
 import devmode as dm
+import mcp_a2a_grant as g
+
+_KEY = b"devmode-test-key"
+_AUM = "sha256:" + "ab" * 32
+
+
+def _grant(*, effect="exec", ops):
+    return g.issue_grant(
+        binding={"spiffe_id": "s", "aum_digest": _AUM, "session_id": "sess_tty1"},
+        capability={"kind": "mcp_tool", "capability_ref": "c", "capability_digest": "sha256:" + "cd" * 32, "effect": effect},
+        decision={"placement": "scheduled", "backend": "k8s", "backend_trust": "trusted"},
+        attestation=g.attestation_bundle(spiffe_id="s", aum_digest=_AUM, tpm_valid=True, cosign_valid=True),
+        constraints={"ops_allow": ops}, signer=g.hmac_signer(_KEY))
+
+
+def test_remote_terminal_authorizes_a_grant_permitting_pty_attach():
+    res = dm.attach_command(namespace="ds-x", pod="pp-abc", grant=_grant(ops=["pty.attach"]),
+                            verifier=g.hmac_verifier(_KEY), session_id="sess_tty1", context="kind-x")
+    assert res["authorized"] is True
+    assert res["command"] == ["kubectl", "--context", "kind-x", "exec", "-it", "-n", "ds-x",
+                              "pod/pp-abc", "-c", "dev", "--", "/bin/sh"]
+
+
+def test_remote_terminal_is_fail_closed_without_pty_attach_or_exec():
+    # grant permits exec but not the pty.attach op
+    r1 = dm.attach_command(namespace="ds-x", pod="p", grant=_grant(ops=["fs.read"]),
+                           verifier=g.hmac_verifier(_KEY), session_id="sess_tty1")
+    assert r1["authorized"] is False
+    # grant is for read, not exec
+    r2 = dm.attach_command(namespace="ds-x", pod="p", grant=_grant(effect="read", ops=["pty.attach"]),
+                           verifier=g.hmac_verifier(_KEY), session_id="sess_tty1")
+    assert r2["authorized"] is False
 
 
 def test_devmode_patch_swaps_in_a_dev_runner_with_a_workspace():

tools/test_login.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python3
+"""Tests for login/session — fail-closed authentication of the front door."""
+from datetime import datetime, timedelta, timezone
+
+import login as lg
+
+KEY = b"login-test-key"
+CREDS = {"alice": "s3cret"}
+
+
+def _check(user, credential):
+    return CREDS.get(user) == credential
+
+
+def test_authenticate_issues_a_session_for_good_credentials():
+    s = lg.authenticate(user="alice", credential="s3cret", key=KEY, tier="pro", credential_check=_check)
+    assert s and s["user"] == "alice" and s["tier"] == "pro" and s["session_id"].startswith("sess_")
+    assert lg.verify_session(s, key=KEY)["valid"] is True
+
+
+def test_bad_credentials_are_rejected_fail_closed():
+    assert lg.authenticate(user="alice", credential="wrong", key=KEY, credential_check=_check) is None
+    assert lg.authenticate(user="mallory", credential="x", key=KEY, credential_check=_check) is None
+
+
+def test_verify_denies_a_tampered_session():
+    s = lg.authenticate(user="alice", credential="s3cret", key=KEY, credential_check=_check)
+    s = {**s, "tier": "enterprise"}  # privilege-escalation attempt
+    r = lg.verify_session(s, key=KEY)
+    assert r["valid"] is False and "tamper" in r["reason"].lower()
+
+
+def test_verify_denies_an_expired_session():
+    now = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    s = lg.issue_session(user="alice", tier="pro", key=KEY, ttl_s=60, now=now)
+    ok = lg.verify_session(s, key=KEY, now=now + timedelta(seconds=30))
+    late = lg.verify_session(s, key=KEY, now=now + timedelta(seconds=120))
+    assert ok["valid"] is True and late["valid"] is False and "expired" in late["reason"]
+
+
+if __name__ == "__main__":
+    import sys
+    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
+    for fn in fns:
+        fn()
+    print(f"ok: {len(fns)} login tests passed")
+    sys.exit(0)

tools/validate.py

@@ -48,6 +48,8 @@
     "tools/inference.py",
     "tools/buildpack.py",
     "tools/provisioning.py",
+    "tools/deploy_flow.py",
+    "tools/login.py",
 ]
 CAPD_KEYS = ("capability_id", "kind", "status", "links", "composes_with", "policy")
 # Every CapD in capd/ must carry the core keys and parse — not just the flagship control-plane one.