From d51c710b964267c05eb5b9bab08d7f5cb6fc1804 Mon Sep 17 00:00:00 2001 From: drknowhow Date: Wed, 23 Sep 2026 18:05:31 -0400 Subject: [PATCH] feat(access): loosen a built-in guard for a while or one session (2.148.0) A lease sets one guard to confirm or allow for 60 s to 8 h, for every session or one, in every project or one, without touching its permanent mode. Rows in ~/.c3/builtin_leases.json count only while the keyring holds the sha256 of the exact row, so forged or edited rows are ignored, and a lease never tightens. The hook and MCP server bind their session id so a per-session lease applies; minting needs the Desk token and the glob. Tests: test_builtin_leases.py; full suite 4595 passed, 7 skipped. Claude-Session: https://claude.ai/code/session_01F6QMHB2mkrDJiQWfC1g9wy --- CHANGELOG.md | 27 ++++ cli/c3.py | 2 +- cli/hook_access_guard.py | 1 + cli/hub_server.py | 57 +++++++- cli/hub_ui/components/hub_access.js | 32 +++++ cli/mcp_server.py | 4 +- pyproject.toml | 2 +- services/access_guard.py | 39 +++++- services/builtin_leases.py | 182 ++++++++++++++++++++++++ tests/test_builtin_leases.py | 210 ++++++++++++++++++++++++++++ 10 files changed, 548 insertions(+), 8 deletions(-) create mode 100644 services/builtin_leases.py create mode 100644 tests/test_builtin_leases.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ec33217..536d2af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,33 @@ All notable changes to Code Context Control (C3) are documented here. The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [2.148.0] - 2026-09-23 + +### Added — loosen a built-in guard for a while, or for one session + +A lease loosens one built-in guard to `confirm` or `allow` for 1 minute to +8 hours, for every session or for one agent session, in one project or all of +them. The permanent mode is never touched, so when the lease ends the guard is +back where it was with nothing to undo. + +- `POST /api/hub/access/builtin/lease` mints one. Leases only loosen, so every + mint needs a C3 Desk client token and the glob typed back. A lease that would + change nothing is refused. +- `POST /api/hub/access/builtin/lease/revoke` ends one early, with no token + needed. +- `GET /api/hub/access/builtin` lists live leases, and the Hub's Built-in + guards panel shows them with a countdown and a Revoke button. + +Leases live in `~/.c3/builtin_leases.json` and are two-key, like modes. A row +counts only while the keyring holds the sha256 of that exact row, so a row +written, edited or copied back by hand is ignored. A lease stricter than the +standing mode is ignored too: leases never tighten. + +A per-session lease needs the evaluating process to know its session. The +PreToolUse hook takes it from the payload, and the MCP server binds it before +each tool call (`access_guard.bind_session`). The Hub, CLI and mobile API have +no session, so only all-session leases apply there. + ## [2.147.0] - 2026-09-23 ### Added — built-in guards can be changed from the Hub and C3 Desk diff --git a/cli/c3.py b/cli/c3.py index 690f4e1..4977b97 100644 --- a/cli/c3.py +++ b/cli/c3.py @@ -91,7 +91,7 @@ # Config CONFIG_DIR = ".c3" CONFIG_FILE = ".c3/config.json" -__version__ = "2.147.0" +__version__ = "2.148.0" # The PreToolUse matcher for native subagent calls (installer and hub migration). AGENT_MATCHER = "Agent|Task" diff --git a/cli/hook_access_guard.py b/cli/hook_access_guard.py index 911fdcb..27bf330 100644 --- a/cli/hook_access_guard.py +++ b/cli/hook_access_guard.py @@ -447,6 +447,7 @@ def run(payload: dict, project_path: Path | None = None, tool_input = payload.get("tool_input", {}) or {} base = str(project_path if project_path is not None else Path.cwd()) session_id = str(payload.get("session_id") or "") + ag.bind_session(session_id) if tool in _WRITE_TOOLS or tool in _READ_TOOLS: fp = _target(tool_input) diff --git a/cli/hub_server.py b/cli/hub_server.py index d0ce930..b978b87 100644 --- a/cli/hub_server.py +++ b/cli/hub_server.py @@ -3896,7 +3896,9 @@ def api_hub_access_builtin(): row["error"] = str(e) if row["modes"] or row["error"]: projects.append(row) - return jsonify({"guards": guards, "projects": projects}) + from services import builtin_leases + return jsonify({"guards": guards, "projects": projects, + "leases": builtin_leases.active()}) @app.route("/api/hub/access/builtin/mode", methods=["POST"]) @@ -3948,6 +3950,59 @@ def api_hub_access_builtin_mode(): return jsonify({**result, "loosened": new < old}) +@app.route("/api/hub/access/builtin/lease", methods=["POST"]) +def api_hub_access_builtin_lease(): + """Loosen a builtin for a while: every session, or one. + + Body: ``{scope, path?, glob, mode, ttl_s, session_id?, confirm}``. A + lease only ever loosens, so it always needs a Desk client token and the + glob typed back, and a lease no looser than the live mode is a 400 + rather than a row that would never apply. ``ttl_s`` is 60 s to 8 h. + """ + from services import builtin_leases + data = request.get_json(silent=True) or {} + scope, project, err = _hub_access_target(data) + if err: + return err + who = _hub_human() + if who is None: + return jsonify({"error": "a temporary built-in change needs C3 Desk", + "needs_human": True}), 403 + canon = access_guard._norm_builtin(data.get("glob")) + mode = str(data.get("mode") or "").strip() + if access_guard._norm_builtin(data.get("confirm")) != canon or not canon: + return jsonify({"error": "a temporary change weakens protection", + "needs_confirmation": True, "confirm_with": canon}), 400 + try: + live = access_guard.effective_builtin_modes(project or ".") \ + if scope == "project" else access_guard._scope_modes("global") + if access_guard.builtin_strictness(canon, mode) >= \ + access_guard.builtin_strictness(canon, live.get(canon, "default")): + return jsonify({"error": f"{canon} is already {live.get(canon, 'default')}" + f" — a {mode} lease would change nothing"}), 400 + row = builtin_leases.mint( + canon, mode, scope=scope, project_path=project or ".", + session_id=str(data.get("session_id") or ""), + ttl_s=data.get("ttl_s"), created_by=who["client_id"]) + except ValueError as exc: + return jsonify({"error": str(exc)}), 400 + _hub_access_audit(f"builtin_lease_{mode}", canon, "builtin", scope, project) + return jsonify({"lease": row}) + + +@app.route("/api/hub/access/builtin/lease/revoke", methods=["POST"]) +def api_hub_access_builtin_lease_revoke(): + """End a lease now. Tightening, so no token. Body: ``{id}``.""" + from services import builtin_leases + lease_id = str((request.get_json(silent=True) or {}).get("id") or "") + gone = builtin_leases.revoke(lease_id) if lease_id else None + if gone is None: + return jsonify({"error": "unknown or already-expired lease"}), 404 + _hub_access_audit("builtin_lease_revoke", gone["glob"], "builtin", + gone["scope"], gone.get("project_path") or None) + return jsonify({"revoked": lease_id}) + + # ── Project management: tasks / milestones / notes (v2.45.0) ────────────── # Direct TaskStore per request (reload-per-op store — no runtime build needed # for mutations); every write audited to the target project's activity log. diff --git a/cli/hub_ui/components/hub_access.js b/cli/hub_ui/components/hub_access.js index 069d75e..8315f3c 100644 --- a/cli/hub_ui/components/hub_access.js +++ b/cli/hub_ui/components/hub_access.js @@ -405,6 +405,7 @@ function AccBuiltinPanel() { catch (e) { setErr(apiErr(e)); } }, []); useEffect(() => { load(); }, [load]); + usePoll(load, 15000); const setMode = async (g, mode) => { setBusy(g.glob); setNeeds(null); @@ -420,7 +421,18 @@ function AccBuiltinPanel() { load(); }; + const revoke = async (l) => { + setBusy(l.id); + try { + await api.post('/api/hub/access/builtin/lease/revoke', { id: l.id }); + notify(`Ended the ${l.mode} lease on ${l.glob}`, 'ok'); + } catch (e) { notify(apiErr(e), 'err'); } + setBusy(''); + load(); + }; + const projects = (data && data.projects) || []; + const leases = (data && data.leases) || []; return (
)} + {leases.length > 0 && ( +
+
+ temporary changes +
+ {leases.map(l => ( +
+ {l.glob} + {l.mode} + + {l.scope === 'project' ? l.project_path : 'all projects'} + {' · '}{l.session_id ? <>session {l.session_id.slice(0, 8)} : 'all sessions'} + {' · '}ends in {accExpiresIn(l.expires_at) || '—'} + +
+ revoke(l)}>Revoke +
+ ))} +
+ )} {projects.length > 0 && (
Set per project:{' '} diff --git a/cli/mcp_server.py b/cli/mcp_server.py index ded762e..4eec5d7 100644 --- a/cli/mcp_server.py +++ b/cli/mcp_server.py @@ -23,9 +23,10 @@ from fastmcp import Context, FastMCP +from cli.tools import _grants from core.host import resolve_host from core.ide import get_profile -from services import delegate_hints +from services import access_guard, delegate_hints from services.auto_memory import AutoMemory from services.context_snapshot import ContextSnapshot from services.runtime import C3Runtime, build_runtime, start_runtime, stop_runtime @@ -508,6 +509,7 @@ def _bg(): def _svc(ctx: Context) -> C3Runtime: rt = ctx.request_context.lifespan_context _ensure_repo_map_once(rt) + access_guard.bind_session(_grants.session_id(rt)) return rt diff --git a/pyproject.toml b/pyproject.toml index 81fbe48..fe4e5cf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "code-context-control" -version = "2.147.0" +version = "2.148.0" description = "Local MCP code-intelligence for AI coding tools: surgical search/read/edit, agent-config version history, path-level access + masking guards, and a multi-project hub." readme = "README.md" requires-python = ">=3.10" diff --git a/services/access_guard.py b/services/access_guard.py index 379e151..e532cea 100644 --- a/services/access_guard.py +++ b/services/access_guard.py @@ -60,6 +60,14 @@ # (or, worse, silently disable one). _ACCESS_KEYRING_SERVICE = "c3-access" +# Temporary and per-session builtin modes (services/builtin_leases.py), in +# ~/.c3. Named here so the hot path can skip the lease module when no file +# exists. +LEASES_FILE = "builtin_leases.json" + +# The agent session this process evaluates for; see bind_session(). +_SESSION_ID = "" + # Mask presets (docs/mask-guard.md §4). Names + param schema live here so the # hook subprocess can validate config without importing the transform engines; # the engines themselves are in services/mask_presets.py. @@ -600,10 +608,33 @@ def effective_builtin_modes(project_path: str = ".") -> dict: """ out = _scope_modes("global") proj_base = _scope_base("project", project_path) - if proj_base is None or proj_base == _global_base(): - return out # the home directory IS the project; one scope, not two - out.update(_scope_modes("project", project_path)) - return out + # When the home directory IS the project there is one scope, not two. + if proj_base is not None and proj_base != _global_base(): + out.update(_scope_modes("project", project_path)) + return _lease_overlay(out, project_path) + + +def bind_session(session_id: str) -> None: + """Name the agent session this process evaluates for, so leases bound to + that session apply. The hook binds per call from its payload; the MCP + server binds before each tool call. Unbound, only all-session leases + apply.""" + global _SESSION_ID + _SESSION_ID = str(session_id or "") + + +def _lease_overlay(modes: dict, project_path: str) -> dict: + """*modes* with live leases applied. A lease only loosens: one that is + stricter than the standing mode is ignored.""" + home = _global_base() + if home is None or not (home / ".c3" / LEASES_FILE).is_file(): + return modes + from services import builtin_leases # noqa: PLC0415 — only when leases exist + for glob, mode in builtin_leases.live_modes(project_path, _SESSION_ID).items(): + if builtin_strictness(glob, mode) < builtin_strictness( + glob, modes.get(glob, "default")): + modes[glob] = mode + return modes def builtin_mode_realms(project_path: str = ".") -> dict: diff --git a/services/builtin_leases.py b/services/builtin_leases.py new file mode 100644 index 0000000..ac95cd9 --- /dev/null +++ b/services/builtin_leases.py @@ -0,0 +1,182 @@ +"""Temporary and per-session modes for Tier-1 builtin guards. + +A lease loosens one builtin to ``confirm`` or ``allow`` until it expires, for +every session (``session_id`` empty) or for one agent session. It never +touches the permanent two-key mode: when a lease lapses the guard is back +where it was, with nothing to undo. + +Leases are two-key like modes. A row in ``~/.c3/builtin_leases.json`` counts +only while the keyring (service ``c3-access``, account ``lease|``) holds +the sha256 of that exact row, so a row written, copied or extended by hand +has no matching attestation and is ignored. Every failure path ignores the +lease, which leaves the guard on. +""" +from __future__ import annotations + +import hashlib +import json +import secrets +from datetime import datetime, timedelta, timezone + +from services import access_guard as ag +from services.atomic_json import write_json_atomic + +LEASE_MODES = ("confirm", "allow") +MIN_TTL_S = 60 +MAX_TTL_S = 8 * 3600 +_LOCK_FILE = ".c3/builtin_leases.lock" + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _parse(value) -> datetime | None: + try: + dt = datetime.fromisoformat(str(value)) + except ValueError: + return None + return dt if dt.tzinfo else None + + +def _store(): + base = ag._global_base() + return base / ".c3" / ag.LEASES_FILE if base is not None else None + + +def _account(lease_id: str) -> str: + return f"lease|{lease_id}" + + +def _digest(row: dict) -> str: + blob = json.dumps(row, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(blob.encode("utf-8")).hexdigest() + + +def _attested(row: dict) -> bool: + try: + import keyring # noqa: PLC0415 — lazy, as in access_guard + return keyring.get_password(ag._ACCESS_KEYRING_SERVICE, + _account(str(row.get("id")))) == _digest(row) + except Exception: + return False + + +def _load() -> list: + path = _store() + if path is None or not path.is_file(): + return [] + try: + rows = json.loads(path.read_text(encoding="utf-8")).get("leases") + except (OSError, ValueError, AttributeError): + return [] + return [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else [] + + +def _live(row: dict, at: datetime) -> bool: + exp = _parse(row.get("expires_at")) + return exp is not None and at < exp + + +def active() -> list: + """Live, attested leases, soonest to expire first.""" + at = _now() + return sorted((r for r in _load() if _live(r, at) and _attested(r)), + key=lambda r: str(r.get("expires_at"))) + + +def live_modes(project_path: str, session_id: str) -> dict: + """``{canonical_glob: mode}`` from the leases that reach this project and + session. ``allow`` wins over ``confirm`` when two leases cover one glob.""" + rows = _load() + if not rows: + return {} + realms = {"global", ag.realm("project", project_path)} + at = _now() + out = {} + for r in rows: + if (r.get("realm") in realms and r.get("mode") in LEASE_MODES + and r.get("session_id", "") in ("", session_id) + and _live(r, at) and _attested(r)): + if out.get(r["glob"]) != "allow": + out[r["glob"]] = r["mode"] + return out + + +def _save(rows: list) -> None: + write_json_atomic(_store(), {"leases": rows}) + + +def _lock(): + from services.override_grants import _Lock # noqa: PLC0415 — off the hot path + return _Lock(ag._global_base(), _LOCK_FILE) + + +def mint(glob, mode: str, *, scope: str, project_path: str = ".", + session_id: str = "", ttl_s: int, created_by: str) -> dict: + """Create a lease and return its row. + + Raises ValueError for a glob that is not Tier 1, a mode other than + confirm/allow, a TTL outside 60 s–8 h, a scope without a realm, or a + keyring that will not attest (attestation is written first, so a lease + that could never count is never stored). + """ + if mode not in LEASE_MODES: + raise ValueError(f"a lease mode must be one of: {', '.join(LEASE_MODES)}") + ag.builtin_strictness(glob, mode) + try: + ttl = int(ttl_s) + except (TypeError, ValueError): + raise ValueError("ttl_s must be a whole number of seconds") from None + if not MIN_TTL_S <= ttl <= MAX_TTL_S: + raise ValueError(f"ttl_s must be between {MIN_TTL_S} and {MAX_TTL_S}") + if scope not in ("global", "project"): + raise ValueError("scope must be 'global' or 'project'") + realm = ag.realm(scope, project_path) + if not realm or _store() is None: + raise ValueError("no home directory to hold the lease") + at = _now() + row = { + "id": "bls_" + secrets.token_hex(6), + "glob": ag._norm_builtin(glob), + "mode": mode, + "scope": scope, + "realm": realm, + "project_path": str(project_path) if scope == "project" else "", + "session_id": str(session_id or ""), + "created_at": at.isoformat(), + "expires_at": (at + timedelta(seconds=ttl)).isoformat(), + "created_by": str(created_by), + } + try: + import keyring # noqa: PLC0415 + keyring.set_password(ag._ACCESS_KEYRING_SERVICE, _account(row["id"]), + _digest(row)) + except Exception: + raise ValueError("keyring unavailable, so the lease cannot be attested " + "and would not take effect") from None + with _lock(): + rows = [r for r in _load() if _live(r, at)] + rows.append(row) + _save(rows) + return row + + +def revoke(lease_id: str) -> dict | None: + """Remove a lease; returns the removed row, or None if it was not live. + Expired rows are swept on the way.""" + if _store() is None: + return None + at = _now() + with _lock(): + rows = _load() + gone = next((r for r in rows if r.get("id") == lease_id), None) + _save([r for r in rows if r is not gone and _live(r, at)]) + if gone is None: + return None + try: + import keyring # noqa: PLC0415 + keyring.set_password(ag._ACCESS_KEYRING_SERVICE, _account(lease_id), "") + except Exception: # a keyring that cannot be written cannot be read either + pass # so a re-added copy of the row stays unattested + return gone if _live(gone, at) else None diff --git a/tests/test_builtin_leases.py b/tests/test_builtin_leases.py new file mode 100644 index 0000000..fec6f19 --- /dev/null +++ b/tests/test_builtin_leases.py @@ -0,0 +1,210 @@ +"""Temporary and per-session builtin guard modes (2.148.0).""" +from __future__ import annotations + +import json +import sys +import tempfile +import types +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT)) + +import cli.hub_server as hub_server # noqa: E402 +from oracle.services import client_tokens # noqa: E402 +from services import access_guard as ag # noqa: E402 +from services import builtin_leases as bl # noqa: E402 + +ENV = "**/.env*" +GIT = "**/.git/**" +DESK = {"Authorization": "Bearer desk-token"} + + +class _FakeKeyring: + def __init__(self): + self.store = {} + + def set_password(self, service, account, value): + self.store[(service, account)] = value + + def get_password(self, service, account): + return self.store.get((service, account)) + + +class _StubPM: + def list_registered(self): + return [] + + +class _Base(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + root = Path(self._tmp.name) + self.home = root / "home" + (self.home / ".c3").mkdir(parents=True) + self.proj = root / "proj" + (self.proj / ".c3").mkdir(parents=True) + self.other = root / "other" + (self.other / ".c3").mkdir(parents=True) + self.keyring = _FakeKeyring() + mod = types.ModuleType("keyring") + mod.set_password = self.keyring.set_password + mod.get_password = self.keyring.get_password + self._patches = [ + mock.patch.dict(sys.modules, {"keyring": mod}), + mock.patch.object(ag, "_global_base", return_value=self.home), + ] + for p in self._patches: + p.start() + ag.bind_session("") + + def tearDown(self): + ag.bind_session("") + for p in reversed(self._patches): + p.stop() + self._tmp.cleanup() + + def env_kind(self, project=None): + project = project or self.proj + denial = ag.check(str(project / ".env"), "read", str(project)) + return denial.kind if denial else "allowed" + + def mint(self, mode="confirm", **kw): + kw.setdefault("scope", "global") + kw.setdefault("ttl_s", 900) + return bl.mint(ENV, mode, created_by="desk-1", **kw) + + def rows(self): + return json.loads((self.home / ".c3" / ag.LEASES_FILE).read_text())["leases"] + + def write_rows(self, rows): + (self.home / ".c3" / ag.LEASES_FILE).write_text(json.dumps({"leases": rows})) + + +class LeaseEvaluation(_Base): + def test_an_all_session_lease_loosens_every_project_until_revoked(self): + self.assertEqual(self.env_kind(), "deny") + row = self.mint("confirm") + self.assertEqual(self.env_kind(), "confirm") + self.assertEqual(self.env_kind(self.other), "confirm") + bl.revoke(row["id"]) + self.assertEqual(self.env_kind(), "deny") + + def test_a_project_lease_stays_in_its_project(self): + self.mint("allow", scope="project", project_path=str(self.proj)) + self.assertEqual(self.env_kind(), "allowed") + self.assertEqual(self.env_kind(self.other), "deny") + + def test_a_session_lease_applies_only_to_the_bound_session(self): + self.mint("allow", session_id="sess-a") + self.assertEqual(self.env_kind(), "deny") + ag.bind_session("sess-b") + self.assertEqual(self.env_kind(), "deny") + ag.bind_session("sess-a") + self.assertEqual(self.env_kind(), "allowed") + + def test_an_expired_lease_does_nothing(self): + row = self.mint("allow") + ends = datetime.fromisoformat(row["expires_at"]) + with mock.patch.object(bl, "_now", return_value=ends): + self.assertEqual(self.env_kind(), "deny") + + def test_a_row_edited_by_hand_loses_its_attestation(self): + self.mint("confirm") + rows = self.rows() + rows[0]["mode"] = "allow" + self.write_rows(rows) + self.assertEqual(self.env_kind(), "deny") + + def test_a_row_written_without_the_keyring_half_is_ignored(self): + at = datetime.now(timezone.utc) + self.write_rows([{"id": "bls_forged", "glob": ENV, "mode": "allow", + "scope": "global", "realm": "global", "project_path": "", + "session_id": "", "created_at": at.isoformat(), + "expires_at": (at + timedelta(hours=1)).isoformat(), + "created_by": "agent"}]) + self.assertEqual(self.env_kind(), "deny") + + def test_a_revoked_row_copied_back_stays_void(self): + row = self.mint("allow") + saved = self.rows() + bl.revoke(row["id"]) + self.write_rows(saved) + self.assertEqual(self.env_kind(), "deny") + + def test_a_lease_never_tightens(self): + ag.set_builtin_mode(GIT, "allow", "global") + bl.mint(GIT, "confirm", scope="global", ttl_s=900, created_by="desk-1") + self.assertIsNone(ag.check(str(self.proj / ".git" / "config"), "write", str(self.proj))) + + def test_mint_refuses_bad_input(self): + for kwargs in ({"ttl_s": 30}, {"ttl_s": 8 * 3600 + 1}, {"ttl_s": "soon"}): + with self.assertRaises(ValueError): + self.mint(**kwargs) + with self.assertRaises(ValueError): + bl.mint(ENV, "deny", scope="global", ttl_s=900, created_by="x") + with self.assertRaises(ValueError): + bl.mint("**/.c3/secrets.enc", "allow", scope="global", ttl_s=900, created_by="x") + + def test_no_file_means_no_keyring_reads(self): + with mock.patch.object(bl, "_attested") as attested: + self.assertEqual(self.env_kind(), "deny") + attested.assert_not_called() + + +class LeaseRoutes(_Base): + def setUp(self): + super().setUp() + principal = {"desk-token": {"kind": "desk", "client_id": "desk-1"}} + self._patches += [ + mock.patch.object(hub_server, "_pm", return_value=_StubPM()), + mock.patch.object(hub_server, "_resolve_project_path", side_effect=Path), + mock.patch.object(client_tokens, "principal_for", side_effect=principal.get), + ] + for p in self._patches[2:]: + p.start() + hub_server.app.config["TESTING"] = True + self.client = hub_server.app.test_client() + + def post(self, url, body, headers=None): + resp = self.client.post(url, json=body, headers=headers or {}) + return resp.status_code, resp.get_json() + + def lease_body(self, **kw): + return {"scope": "global", "glob": ENV, "mode": "confirm", "ttl_s": 900, + "confirm": ENV, **kw} + + def test_minting_needs_desk(self): + status, body = self.post("/api/hub/access/builtin/lease", self.lease_body()) + self.assertEqual(status, 403) + self.assertTrue(body["needs_human"]) + + def test_minting_needs_the_glob_retyped(self): + status, _ = self.post("/api/hub/access/builtin/lease", + self.lease_body(confirm="x"), DESK) + self.assertEqual(status, 400) + + def test_a_lease_that_would_change_nothing_is_refused(self): + ag.set_builtin_mode(ENV, "allow", "global") + status, _ = self.post("/api/hub/access/builtin/lease", self.lease_body(), DESK) + self.assertEqual(status, 400) + + def test_mint_list_revoke_round_trip(self): + status, body = self.post("/api/hub/access/builtin/lease", + self.lease_body(session_id="sess-a"), DESK) + self.assertEqual(status, 200) + lease_id = body["lease"]["id"] + listed = self.client.get("/api/hub/access/builtin").get_json()["leases"] + self.assertEqual([row["id"] for row in listed], [lease_id]) + self.assertEqual(listed[0]["created_by"], "desk-1") + status, _ = self.post("/api/hub/access/builtin/lease/revoke", {"id": lease_id}) + self.assertEqual(status, 200) + status, _ = self.post("/api/hub/access/builtin/lease/revoke", {"id": lease_id}) + self.assertEqual(status, 404) + + +if __name__ == "__main__": + unittest.main()