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()