Resolve security-review BLOCK: CI-guard bypasses, denylist parity, force-resume

Addresses the confirmed findings from /sh-security-review + the GPT-4.1
cross-review of the Plane-2 scaffold. Full suite: 589 passed; ruff clean.

FIXED (proven-exploitable):
- CI-guard denylist bypass (HIGH): Python fnmatch '**/' is non-recursive, so
  root-level template.yaml/*.tf/cdk.json/*.pem/*.key/*-stack.* evaded the
  trust-control surface. Replaced fnmatch with a recursive, case-insensitive
  glob->regex matcher. (verified: fnmatch('template.yaml','**/template.yaml')==False)
- CI-guard scope bypass (HIGH): a '**' declared_scope made every path in-scope.
  Scope is now concrete-prefix confinement (reduces a glob to its leading
  metacharacter-free segments; '**' -> empty -> dropped -> unscoped reject).
- Box-side vs CI denylist divergence (MED): builders.py _DENY_PATTERNS now covers
  Terraform, *.pem/*.key, CDK stack files, .github/actions, *iam*, bare policy*.json
  (case-insensitive), matching the CI surface.
- force-resume was backwards (MED): it superseded the answered row recovery
  resumes from, making a stuck task permanently un-resumable while printing
  success. Now re-opens an EXPIRED (parked) question via a new reopen_question
  CAS helper; never supersedes an answered row; honest exit codes.
- operator attribution (MED): run-team.py --operator defaulted to "" -> now the
  OS login, so destructive actions are always attributable.
- audit-log append race (MED): replaced read-modify-rewrite (lost records under
  concurrent operators) with an O_APPEND single-line write, mode 600 enforced.
- lstrip("ab/") path-mangling in the symlink error path -> regex prefix strip.

Regression tests added across test_ci_gate_workflow / test_builders / test_run_team
/ test_schema. Design-level findings (resume-worker durability, egress breadth,
answered_at ordering, DB-swap TOCTOU, diff-hash threat-model) are pre-deployment
/ P1-build-proper and recorded with written justification in
agent-team/.security-review/suppressions.json; CI README diff-hash wording made
honest.
This commit is contained in:
Adam Moussa 2026-06-17 15:15:50 -04:00
parent 0eae5dbfc3
commit 721cec5315
10 changed files with 419 additions and 82 deletions

View file

@ -0,0 +1,25 @@
{
"_comment": "Written justifications for sh-security-review (Path A) + GPT-4.1 cross-review findings on the R720 agent-team Plane-2 scaffold that are deliberately NOT fixed in this pre-deployment commit. Per CLAUDE.md: a confirmed critical/high is either fixed or suppressed with a written justification. Every item here is design-level / deferred-to-P1-build and is NOT live-exploitable because nothing in agent-team/ is provisioned, scheduled, or enabled. The proven-exploitable HIGHs (CI-guard fnmatch '**/' denylist bypass and the scope '**' bypass) were FIXED, not suppressed (see ci/agent-team-apply-verify.yml + tests/test_ci_gate_workflow.py).",
"suppressions": [
{
"id": "LOGIC-01/02/03-resume-worker-durability",
"justification": "The resume worker's no-double-apply / single-flight currently rests on an in-process threading.Lock + a turn-integer guard, and an 'answered' ledger row has no terminal 'resumed' transition, so a restart sweep re-enqueues it (correctness then relies on the turn guard). This is a real durability gap, but: (1) it is PRE-DEPLOYMENT scaffolding — no resume worker, responder, or scheduler runs anywhere, so it cannot be triggered in the current state; (2) the design (§3.3.1) explicitly frames the durable, cross-process single-flight resume worker as P1-build-proper. FIX TRACKED FOR P1 BUILD: add a terminal 'resumed' state to QUESTION_STATES, flip 'answered'->'resumed' via the BEGIN IMMEDIATE compare-and-set inside the resume success path (making the CAS the cross-process single-flight token), have the turn guard compare question_id identity (not just the turn integer), and filter recovery selects to exclude resumed rows. Until the worker is built and enabled, no merge of live resume behavior occurs."
},
{
"id": "SC-01-diff-hash-threat-model",
"justification": "The unkeyed SHA-256 diff-integrity hash defends in-transit corruption/substitution between the trusted dispatcher and CI, NOT a compromised box (the box is in the trusted computing base for what it proposes). This is already stated honestly in the design doc §3.3.2 ('Threat-model honesty (the diff-hash does not cover a compromised box)') and the ci/README.md threat-model section has been aligned to match. The real backstops against a bad box are the credential-less untrusted build job, the trust-control-surface denylist, and the mandatory human review of the draft PR + required checks. A keyed/attested binding to the approved-plan record is a P1/P3 hardening, not a current vuln."
},
{
"id": "ATCI-003-004-build-test-egress-breadth",
"justification": "The build-test egress allowlist includes GitHub API / githubusercontent / a wildcard actions host beyond the strict minimum. MEDIUM/LOW: the primary §3.3.2 mitigation is that the untrusted build-test job is credential-less (no secrets, no OIDC, contents:read), so a successful exfil yields nothing of value. The workflow is authored-but-disabled scaffolding; the file already carries a DEPLOY marker requiring the allowlist be trimmed to exactly the target repo's registries before the workflow is enabled. Over-trimming now risks breaking checkout/artifact flows in a workflow that does not yet run. TRACKED as a deploy-time hardening item."
},
{
"id": "LOGIC-06-answered_at-stamp-before-lock",
"justification": "answer_question computes answered_at before BEGIN IMMEDIATE acquires the write lock, so under contention the persisted timestamp can invert commit order, which recovery uses to order cross-question replay. LOW: cross-question replay ordering does not affect P1 correctness (each thread_id is independent and per-thread ordering is preserved by the turn sequence). PRE-DEPLOYMENT. TRACKED for P1: stamp inside the transaction (SQLite strftime in the UPDATE) or order recovery by a monotonic rowid instead of answered_at."
},
{
"id": "XREVIEW-7-cas-db-identity-toctou",
"justification": "_compare_and_set derives the backing DB file and opens a private connection per call; if the DB file were swapped/moved between connect() and the CAS, a stale file could be resolved. LOW/edge: requires an attacker with local filesystem write to swap the durable store mid-operation, at which point they already control the ledger directly (the deployment-model finding XREVIEW-9 / file-permissions, which is an OS-level access-control concern documented for the runbook). PRE-DEPLOYMENT; the box is read-only with mode-600 state per the design. TRACKED for the operational runbook (filesystem permissions + integrity) rather than a code change."
}
]
}

View file

@ -39,6 +39,7 @@ __all__ = [
"expire_question", "expire_question",
"init_db", "init_db",
"migrate", "migrate",
"reopen_question",
"supersede_question", "supersede_question",
] ]
@ -267,6 +268,35 @@ def expire_question(
) )
def reopen_question(
conn: sqlite3.Connection,
*,
question_id: str,
deadline_at: str | None = None,
) -> bool:
"""Un-park: flip an ``expired`` question back to ``open`` (operator action).
The §6.6 operator force-resume path for a parked task whose clarifier
question expired with no answer: re-open it so the normal delivery → answer →
resume flow can proceed, instead of destructively superseding it (which would
remove it from the recovery sweep's reach). Same compare-and-set discipline —
only an ``expired`` row is reopened; an already-answered/open/superseded row
loses the CAS and is untouched. ``deadline_at`` sets a fresh window (``NULL``
means no deadline until one is set, so it will not immediately re-expire).
Returns ``True`` if this call reopened the question.
"""
return _compare_and_set(
conn,
sql=(
"UPDATE pending_questions "
"SET status='open', deadline_at=?, channel_ref=NULL, "
"answer_json=NULL, answered_via=NULL, answered_at=NULL "
"WHERE question_id=? AND status='expired'"
),
params=(deadline_at, question_id),
)
def supersede_question( def supersede_question(
conn: sqlite3.Connection, conn: sqlite3.Connection,
*, *,

View file

@ -104,8 +104,24 @@ _DENY_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
r"(^|/)cdk\.json$" r"(^|/)cdk\.json$"
r"|(^|/).+\.(iam|policy)\.(json|ya?ml)$" r"|(^|/).+\.(iam|policy)\.(json|ya?ml)$"
r"|(^|/)(iam|policies|policy)/.+\.(json|ya?ml)$" r"|(^|/)(iam|policies|policy)/.+\.(json|ya?ml)$"
# Parity with the CI-side denylist (case-insensitive): bare policy
# docs, any path naming "iam", Terraform, and CDK stack files —
# these were box-side gaps a diff could use to skip the cross-review.
r"|(^|/)policy[^/]*\.json$"
r"|(^|/)[^/]*iam[^/]*$"
r"|\.tf$"
r"|(^|/).+[-_]stack\.(ts|py)$",
re.IGNORECASE,
), ),
"modifies IAM/policy IaC", "modifies IAM/policy/Terraform/CDK IaC",
),
(
re.compile(r"^\.github/actions/.+", re.IGNORECASE),
"modifies a composite GitHub Action (.github/actions/**)",
),
(
re.compile(r"\.(pem|key)$", re.IGNORECASE),
"modifies key material (*.pem / *.key)",
), ),
) )

View file

@ -51,9 +51,14 @@ implements all five boundaries:
GPT cross-review, never auto-built. GPT cross-review, never auto-built.
3. **Diff integrity, box → CI.** The builder records the candidate diff's sha256 3. **Diff integrity, box → CI.** The builder records the candidate diff's sha256
in the task ledger (foundation `agent_team.state_store` content-hash idiom). in the task ledger (foundation `agent_team.state_store` content-hash idiom).
CI **re-hashes the diff and verifies it equals the ledger-recorded hash CI **re-hashes the diff and verifies it equals the recorded hash before
before applying** (in both `guard` and again pre-apply in `build-test`). A applying** (in both `guard` and again pre-apply in `build-test`). Precisely:
tampered or substituted diff fails the hash check. this is an **unkeyed** hash that binds *the bytes CI applies* to *the hash the
dispatcher recorded* — it detects accidental corruption or substitution of the
artifact **in transit** between the trusted dispatcher and CI. It does **not**
prove the diff matches the approved plan, and it cannot defend a compromised
box that generates both the diff and its hash (see "Threat-model honesty"
below). A keyed/attested binding to the approval record is a later hardening.
4. **Pure-code pass/fail gate over authenticated results.** Mirroring secrev's 4. **Pure-code pass/fail gate over authenticated results.** Mirroring secrev's
"one pure-code script owns the block decision," the `gate-and-pr` gate reads "one pure-code script owns the block decision," the `gate-and-pr` gate reads
**only** the authenticated `needs.*.result` job conclusions (GitHub-controlled, **only** the authenticated `needs.*.result` job conclusions (GitHub-controlled,

View file

@ -134,7 +134,6 @@ jobs:
python3 - <<'PY' python3 - <<'PY'
from __future__ import annotations from __future__ import annotations
import fnmatch
import hashlib import hashlib
import os import os
import posixpath import posixpath
@ -244,33 +243,98 @@ jobs:
try: try:
path_c = canonical(cur_path) if cur_path else "<unknown>" path_c = canonical(cur_path) if cur_path else "<unknown>"
except ValueError: except ValueError:
path_c = (cur_path or "<unknown>").lstrip("ab/") # canonical() rejected the path (absolute/escaping); report
# it with only the git a//b/ PREFIX removed for the error
# message (re.sub, not str.lstrip which strips a char set).
path_c = re.sub(r"^[ab]/", "", cur_path or "<unknown>")
additions.append((path_c, line[1:].strip())) additions.append((path_c, line[1:].strip()))
pending = False pending = False
return additions return additions
_GLOB_META = set("*?[]")
_GLOB_RE_CACHE: dict[str, "re.Pattern[str]"] = {}
def _glob_to_regex(glob: str) -> "re.Pattern[str]":
"""Compile a gitignore-style glob to a '/'-aware, case-insensitive regex.
Python's ``fnmatch`` does NOT implement recursive ``**`` (it treats it
as a single ``*`` that already spans ``/``), so ``**/template.yaml``
fails to match a repo-ROOT ``template.yaml`` — a denylist bypass for
exactly the IaC/secret families boundary 2 must catch. This translates
``**/`` to "any depth INCLUDING zero", ``**`` to ".*", ``*`` to a single
non-slash run, ``?`` to one non-slash char, and matches case-
insensitively (POSIX runners are case-sensitive, but a case variant of a
trust-control filename must not slip the gate).
"""
cached = _GLOB_RE_CACHE.get(glob)
if cached is not None:
return cached
out: list[str] = []
i, n = 0, len(glob)
while i < n:
if glob[i : i + 3] == "**/":
out.append(r"(?:.*/)?")
i += 3
elif glob[i : i + 2] == "**":
out.append(r".*")
i += 2
elif glob[i] == "*":
out.append(r"[^/]*")
i += 1
elif glob[i] == "?":
out.append(r"[^/]")
i += 1
else:
out.append(re.escape(glob[i]))
i += 1
pat = re.compile("^" + "".join(out) + "$", re.IGNORECASE)
_GLOB_RE_CACHE[glob] = pat
return pat
def denied(path: str) -> bool: def denied(path: str) -> bool:
"""True if ``path`` matches any trust-control denylist glob.""" """True if ``path`` is on the trust-control denylist (recursive, case-insensitive)."""
return any(fnmatch.fnmatch(path, g) for g in DENY_GLOBS) return any(_glob_to_regex(g).match(path) for g in DENY_GLOBS)
def _scope_prefix(entry: str) -> str | None:
"""Reduce a canonicalized scope entry to a concrete dir/file prefix.
Declared scope is *confinement*, not a pattern that may widen coverage.
``fnmatch``-ing scope let a single ``**`` (or ``*``) entry match the
whole tree, collapsing boundary 2b to a no-op. Instead we take the
leading path segments up to the first glob metacharacter and prefix
-match against them (mirrors the box-side ``_in_scope``). A scope that
begins with a metacharacter reduces to the empty (repo-root) prefix and
is dropped, so it can never widen to everything.
"""
keep: list[str] = []
for part in entry.split("/"):
if any(c in _GLOB_META for c in part):
break
keep.append(part)
prefix = "/".join(keep)
return prefix or None
def safe_scope(scope: list[str]) -> list[str]: def safe_scope(scope: list[str]) -> list[str]:
"""Canonicalize declared-scope globs, dropping any that escape root. """Canonicalize scope into concrete path prefixes; drop escaping/empty.
An absolute or parent-escaping scope entry is discarded rather than An absolute or parent-escaping entry is discarded (canonical raises),
trusted, so a malformed scope can only SHRINK what is allowed, never and a glob that reduces to the repo root is dropped, so a malformed or
widen it past the repo root (mirrors the box-side normalizer). over-broad scope can only SHRINK what is allowed, never widen it.
""" """
safe: list[str] = [] safe: list[str] = []
for g in scope: for g in scope:
try: try:
safe.append(canonical(g)) canon = canonical(g)
except ValueError: except ValueError:
continue continue
prefix = _scope_prefix(canon)
if prefix is not None and prefix not in safe:
safe.append(prefix)
return safe return safe
def in_scope(path: str, scope: list[str]) -> bool: def in_scope(path: str, scope: list[str]) -> bool:
"""True if ``path`` is covered by the declared-scope globs.""" """True if ``path`` is at or under one of the declared scope prefixes."""
return any(fnmatch.fnmatch(path, g) for g in scope) return any(path == entry or path.startswith(entry + "/") for entry in scope)
def main() -> int: def main() -> int:
diff_path = os.environ["DIFF_PATH"] diff_path = os.environ["DIFF_PATH"]

View file

@ -46,7 +46,9 @@ compare-and-set lost the race), ``2`` usage error (argparse).
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import getpass
import json import json
import os
import sqlite3 import sqlite3
import sys import sys
from datetime import datetime, timezone from datetime import datetime, timezone
@ -67,9 +69,9 @@ from agent_team.db.schema import ( # noqa: E402 (path bootstrap must precede)
connect, connect,
expire_question, expire_question,
init_db, init_db,
reopen_question,
supersede_question, supersede_question,
) )
from agent_team.state_store import atomic_write # noqa: E402
__all__ = [ __all__ = [
"build_parser", "build_parser",
@ -112,29 +114,48 @@ def _utc_now_iso() -> str:
return datetime.now(timezone.utc).isoformat() return datetime.now(timezone.utc).isoformat()
def _default_operator() -> str:
"""Best-effort OS login for audit attribution (never an empty string).
A previous empty default left destructive actions non-attributable (the
audit record named no one). Defaulting to the OS login keeps the §3.3.1
"audit-logged AND attributable" guarantee even when --operator is omitted;
falls back to "unknown" only if the login cannot be resolved.
"""
try:
user = getpass.getuser()
except Exception: # noqa: BLE001 - getuser can raise on odd environments
return "unknown"
return user or "unknown"
def _row_to_dict(row: sqlite3.Row) -> dict[str, Any]: def _row_to_dict(row: sqlite3.Row) -> dict[str, Any]:
"""Project a ``pending_questions`` row to a plain dict for display.""" """Project a ``pending_questions`` row to a plain dict for display."""
return {col: row[col] for col in _QUESTION_COLUMNS if col in row.keys()} return {col: row[col] for col in _QUESTION_COLUMNS if col in row.keys()}
def _append_audit(audit_log: Path, entry: dict[str, Any]) -> None: def _append_audit(audit_log: Path, entry: dict[str, Any]) -> None:
"""Append one JSON audit record crash-safely (read-modify-rewrite). """Append one JSON audit record with an atomic ``O_APPEND`` single write.
Destructive actions (§3.3.1) must leave an attributable trail. We keep an Destructive actions (§3.3.1) must leave an attributable trail. A previous
append-only JSONL file written through the foundation's read-modify-rewrite design lost records under concurrent operators (two
:func:`atomic_write` (write-temp → fsync → rename) so a crash mid-append processes each read the same bytes and the last rewrite wins). Instead each
never tears the log. The file is small (operator actions only), so reading record is one line written with ``O_APPEND``: the kernel serializes the
it back and rewriting it atomically is acceptable and keeps the durability append and a write below ``PIPE_BUF`` is atomic on POSIX, so concurrent
guarantee without introducing a second write primitive. appends never clobber each other. The file is created mode ``0600``
(operator identity / action content is sensitive) and re-chmod'd in case it
pre-existed wider. A failure here raises ``OSError`` BEFORE any ledger
mutation, preserving the audit-before-mutate guarantee.
""" """
audit_log = Path(audit_log) audit_log = Path(audit_log)
existing = b"" audit_log.parent.mkdir(parents=True, exist_ok=True)
if audit_log.exists(): line = (json.dumps(entry, sort_keys=True) + "\n").encode("utf-8")
existing = audit_log.read_bytes() fd = os.open(str(audit_log), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
if existing and not existing.endswith(b"\n"): try:
existing += b"\n" os.write(fd, line)
line = json.dumps(entry, sort_keys=True).encode("utf-8") + b"\n" finally:
atomic_write(audit_log, existing + line) os.close(fd)
os.chmod(audit_log, 0o600)
def _audit_attempt( def _audit_attempt(
@ -418,45 +439,13 @@ def _cmd_answer(args: argparse.Namespace, *, out: Any) -> int:
def _cmd_supersede(args: argparse.Namespace, *, out: Any) -> int: def _cmd_supersede(args: argparse.Namespace, *, out: Any) -> int:
"""Mark a stale question ``superseded`` (destructive; audit-logged).""" """Mark a stale ``open``/``answered`` question ``superseded`` (destructive)."""
return _supersede_impl(args, out=out, action="supersede") _require_confirm("supersede", confirm=args.confirm)
def _cmd_force_resume(args: argparse.Namespace, *, out: Any) -> int:
"""Force-resume a parked task's question (destructive; audit-logged).
The design-named operator verb (§3.3.1 "force-resume" / §6.6 "an operator
can force-resume ... a parked task via the CLI"). The resume worker proper
runs in a later phase; here we mark the stale question ``superseded`` via the
committed turn-guard helper so a redelivered/stale resume job is skipped
idempotently, and record the resume intent durably in the audit trail for
the worker's next sweep to consume.
"""
return _supersede_impl(
args, out=out, action="force-resume", detail={"resume_requested": True}
)
def _supersede_impl(
args: argparse.Namespace,
*,
out: Any,
action: str,
detail: dict[str, Any] | None = None,
) -> int:
"""Shared body for ``supersede`` and ``force-resume``.
Both flip a stale ``open``/``answered`` question to ``superseded`` via the
committed compare-and-set; ``force-resume`` additionally records resume
intent. Audits the attempt before mutating and the outcome after.
"""
_require_confirm(action, confirm=args.confirm)
_audit_attempt( _audit_attempt(
args.audit_log, args.audit_log,
action, "supersede",
question_id=args.question_id, question_id=args.question_id,
operator=args.operator, operator=args.operator,
detail=detail,
) )
conn = connect(args.db) conn = connect(args.db)
try: try:
@ -465,22 +454,11 @@ def _supersede_impl(
conn.close() conn.close()
_audit_outcome( _audit_outcome(
args.audit_log, args.audit_log,
action, "supersede",
question_id=args.question_id, question_id=args.question_id,
operator=args.operator, operator=args.operator,
applied=changed, applied=changed,
detail=detail,
) )
if action == "force-resume":
# Force-resume succeeds as an operator intent even if there was no stale
# row to supersede (the task may already be past the question); the
# intent is recorded for the resume worker either way.
print(
f"recorded force-resume intent for {args.question_id}"
+ ("" if changed else " (no open/answered question to supersede)"),
file=out,
)
return 0
if not changed: if not changed:
print( print(
f"supersede no-op: question {args.question_id} was not " f"supersede no-op: question {args.question_id} was not "
@ -492,6 +470,72 @@ def _supersede_impl(
return 0 return 0
def _cmd_force_resume(args: argparse.Namespace, *, out: Any) -> int:
"""Force-resume a parked task's question (destructive; audit-logged).
The design-named operator verb (§3.3.1 / §6.6 "an operator can force-resume
... a parked task via the CLI"). A task parks when its clarifier question
EXPIRES with no answer, so the un-park action is to RE-OPEN that expired
question (:func:`agent_team.db.schema.reopen_question`) so the normal
delivery → answer → resume flow can proceed.
Crucially this does NOT ``supersede`` the row: superseding an ``answered``
row would flip it out of the state the recovery sweep resumes from, making a
stuck-but-answered task permanently un-resumable — the opposite of
force-resume. So:
* ``expired`` (the parked case) → reopened; returns 0.
* ``answered`` (answered but not yet resumed) → already eligible for the
recovery resume sweep; intent is recorded and we report that, no mutation.
* ``open`` / ``superseded`` / absent → nothing to force; reported as a no-op.
"""
_require_confirm("force-resume", confirm=args.confirm)
_audit_attempt(
args.audit_log,
"force-resume",
question_id=args.question_id,
operator=args.operator,
detail={"resume_requested": True},
)
conn = connect(args.db)
try:
row = _fetch_question(conn, args.question_id)
status = None if row is None else row["status"]
reopened = False
if status == "expired":
reopened = reopen_question(conn, question_id=args.question_id)
finally:
conn.close()
_audit_outcome(
args.audit_log,
"force-resume",
question_id=args.question_id,
operator=args.operator,
applied=reopened,
detail={"resume_requested": True, "prior_status": status},
)
if reopened:
print(
f"force-resume: reopened expired question {args.question_id}; "
"it will be re-delivered for an answer",
file=out,
)
return 0
if status == "answered":
print(
f"force-resume: question {args.question_id} is answered and pending "
"resume; the recovery sweep will resume it (intent recorded)",
file=out,
)
return 0
print(
f"force-resume no-op: question {args.question_id} "
f"({'absent' if status is None else f'status={status}'}) is not parked",
file=sys.stderr,
)
return 1
# Statuses an operator treats as "parked context": a task whose only pending # Statuses an operator treats as "parked context": a task whose only pending
# question is no longer open may be parked (answered-but-unresumed, expired, or # question is no longer open may be parked (answered-but-unresumed, expired, or
# superseded). ``open`` is excluded — that is the live-waiting view (default # superseded). ``open`` is excluded — that is the live-waiting view (default
@ -526,8 +570,9 @@ def build_parser() -> argparse.ArgumentParser:
) )
parser.add_argument( parser.add_argument(
"--operator", "--operator",
default="", default=_default_operator(),
help="operator identity recorded in the audit log for destructive actions", help="operator identity recorded in the audit log for destructive actions "
"(defaults to the OS login so the trail is always attributable)",
) )
sub = parser.add_subparsers(dest="command", required=True) sub = parser.add_subparsers(dest="command", required=True)

View file

@ -300,6 +300,36 @@ def test_out_of_scope_delete_is_flagged() -> None:
assert "scope" in v[0].reason assert "scope" in v[0].reason
# --------------------------------------------------------------------------- #
# Box-side / CI-side denylist parity (security-review: box-side was narrower)
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"path",
[
"infra/main.tf",
"infra/app-stack.ts",
"infra/app_stack.py",
"secrets/deploy.pem",
"signing.key",
"config/policy.json",
"roles/myiam.json",
".github/actions/build/action.yml",
],
)
def test_box_side_denylist_covers_ci_surface(path: str) -> None:
# These IaC/key families were caught by the CI guard but slipped the box-side
# scan. The box is the first backstop; it must reject them too (no auto-build).
diff = (
f"diff --git a/{path} b/{path}\n--- a/{path}\n+++ b/{path}\n"
"@@ -1 +1 @@\n-x\n+y\n"
)
scope = [path.split("/")[0], "."]
v = scan_trust_control_surface(diff, scope=scope)
assert v, f"expected {path} to be denied box-side"
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# Unsafe paths (absolute / parent-escaping / indirection) # Unsafe paths (absolute / parent-escaping / indirection)
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #

View file

@ -139,3 +139,58 @@ def test_escaping_scope_entries_are_dropped(guard_script: Path, tmp_path: Path)
# A parent-escaping scope entry must not widen coverage; it is dropped, so a # A parent-escaping scope entry must not widen coverage; it is dropped, so a
# diff under it is treated as unscoped. # diff under it is treated as unscoped.
assert _run_guard(guard_script, tmp_path, CLEAN, "../../etc") == 5 assert _run_guard(guard_script, tmp_path, CLEAN, "../../etc") == 5
# --- security-review regressions: denylist & scope matcher (was fnmatch) ----
def _modify(path: str) -> str:
return f"diff --git a/{path} b/{path}\n--- a/{path}\n+++ b/{path}\n@@ -1 +1 @@\n-x\n+y\n"
@pytest.mark.parametrize(
"root_path",
[
"template.yaml",
"main.tf",
"cdk.json",
"policy.json",
"id.pem",
"signing.key",
"app-stack.ts",
"infra_stack.py",
],
)
def test_root_level_iac_is_denied(
guard_script: Path, tmp_path: Path, root_path: str
) -> None:
# Regression: Python fnmatch '**/' is non-recursive, so root-level IaC/secret
# files slipped the denylist. The glob->regex matcher must reject them (exit 4).
assert _run_guard(guard_script, tmp_path, _modify(root_path), ".") == 4
@pytest.mark.parametrize(
"cased_path", ["Template.YAML", "Main.TF", ".github/Workflows/ci.yml"]
)
def test_denylist_is_case_insensitive(
guard_script: Path, tmp_path: Path, cased_path: str
) -> None:
# A case variant of a trust-control filename must not evade the gate.
assert _run_guard(guard_script, tmp_path, _modify(cased_path), ".") == 4
def test_scope_double_star_cannot_widen_to_whole_tree(
guard_script: Path, tmp_path: Path
) -> None:
# Regression: a '**' scope entry made in_scope() true for every path. It must
# reduce to the empty (root) prefix and be dropped -> unscoped (exit 5).
assert _run_guard(guard_script, tmp_path, _modify("any/deep/file.py"), "**") == 5
def test_scope_glob_reduces_to_concrete_prefix(
guard_script: Path, tmp_path: Path
) -> None:
# A legitimate 'src/**' scope still confines to the src/ prefix: in-scope
# passes, a sibling path is rejected.
assert _run_guard(guard_script, tmp_path, _modify("src/app.py"), "src/**") == 0
assert _run_guard(guard_script, tmp_path, _modify("other/app.py"), "src/**") == 6

View file

@ -540,22 +540,45 @@ def test_redeliver_non_open_is_noop_returns_1(
assert entries[-1]["applied"] is False assert entries[-1]["applied"] is False
def test_force_resume_supersedes_and_records_intent( def test_force_resume_reopens_expired_parked_question(
cli: ModuleType, db_path: Path, audit_log: Path cli: ModuleType, db_path: Path, audit_log: Path
) -> None: ) -> None:
_insert_question(db_path, question_id="q1", status="open") # The parked case: an expired question is RE-OPENED so it can be answered,
# NOT superseded (superseding would make it permanently un-resumable).
_insert_question(db_path, question_id="q1", status="expired")
code, out = _run( code, out = _run(
cli, db_path, audit_log, "--operator", "adam", "force-resume", "q1", "--confirm" cli, db_path, audit_log, "--operator", "adam", "force-resume", "q1", "--confirm"
) )
assert code == 0 assert code == 0
assert _status_of(db_path, "q1") == "superseded" assert _status_of(db_path, "q1") == "open" # reopened, not superseded
entries = [json.loads(line) for line in audit_log.read_text().splitlines()] entries = [json.loads(line) for line in audit_log.read_text().splitlines()]
assert [e["phase"] for e in entries] == ["attempt", "outcome"] assert [e["phase"] for e in entries] == ["attempt", "outcome"]
assert entries[-1]["action"] == "force-resume" assert entries[-1]["action"] == "force-resume"
assert entries[-1]["resume_requested"] is True assert entries[-1]["resume_requested"] is True
assert entries[-1]["applied"] is True
assert entries[-1]["operator"] == "adam" assert entries[-1]["operator"] == "adam"
def test_force_resume_does_not_supersede_answered_row(
cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
# Regression: force-resume must NOT flip an answered-but-unresumed row out of
# the state the recovery sweep resumes from. It stays 'answered'.
_insert_question(db_path, question_id="q1", status="answered")
code, _ = _run(cli, db_path, audit_log, "force-resume", "q1", "--confirm")
assert code == 0
assert _status_of(db_path, "q1") == "answered" # untouched, still resumable
def test_force_resume_on_open_is_noop(
cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
_insert_question(db_path, question_id="q1", status="open")
code, _ = _run(cli, db_path, audit_log, "force-resume", "q1", "--confirm")
assert code == 1 # an open (not parked) question has nothing to force
assert _status_of(db_path, "q1") == "open"
def test_force_resume_without_confirm_refuses( def test_force_resume_without_confirm_refuses(
cli: ModuleType, db_path: Path, audit_log: Path cli: ModuleType, db_path: Path, audit_log: Path
) -> None: ) -> None:
@ -570,6 +593,19 @@ def test_force_resume_is_in_destructive_set(cli: ModuleType) -> None:
assert "force-resume" in cli._DESTRUCTIVE_ACTIONS assert "force-resume" in cli._DESTRUCTIVE_ACTIONS
def test_operator_defaults_to_os_login_not_empty(
cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
# AUTHZ regression: --operator defaulted to "" → non-attributable audit.
# Omitting it must record a real (non-empty) operator identity.
_insert_question(db_path, question_id="q1", status="open")
code, _ = _run(cli, db_path, audit_log, "expire", "q1", "--confirm")
assert code == 0
entries = [json.loads(line) for line in audit_log.read_text().splitlines()]
assert entries[-1]["operator"] # non-empty
assert cli._default_operator() # helper never returns empty
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# Audit-before-mutate: an unwritable audit path aborts BEFORE the ledger mutates # Audit-before-mutate: an unwritable audit path aborts BEFORE the ledger mutates
# (regression: previously the row was mutated, then the audit append crashed, # (regression: previously the row was mutated, then the audit append crashed,

View file

@ -18,6 +18,7 @@ from agent_team.db.schema import (
expire_question, expire_question,
init_db, init_db,
migrate, migrate,
reopen_question,
supersede_question, supersede_question,
) )
@ -193,6 +194,36 @@ def test_supersede_open_or_answered(tmp_path: Path) -> None:
conn.close() conn.close()
def test_reopen_question_unparks_expired_only(tmp_path: Path) -> None:
db = tmp_path / "db.sqlite"
init_db(db)
conn = connect(db)
try:
_insert_open_question(conn, "exp")
_insert_open_question(conn, "ans")
assert expire_question(conn, question_id="exp") is True
assert answer_question(
conn, question_id="ans", answer_json="{}", answered_via="t"
)
# Expired -> reopened.
assert reopen_question(conn, question_id="exp") is True
row = conn.execute(
"SELECT status, deadline_at FROM pending_questions WHERE question_id='exp'"
).fetchone()
assert row["status"] == "open"
assert row["deadline_at"] is None # no deadline until one is set
# Answered row is NOT reopenable (only expired rows are).
assert reopen_question(conn, question_id="ans") is False
assert (
conn.execute(
"SELECT status FROM pending_questions WHERE question_id='ans'"
).fetchone()["status"]
== "answered"
)
finally:
conn.close()
def test_concurrent_answers_single_winner(tmp_path: Path) -> None: def test_concurrent_answers_single_winner(tmp_path: Path) -> None:
"""Two threads racing to answer the same open question: exactly one wins.""" """Two threads racing to answer the same open question: exactly one wins."""
db = tmp_path / "db.sqlite" db = tmp_path / "db.sqlite"