Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions docs/chapters/architecture.texi
Original file line number Diff line number Diff line change
Expand Up @@ -637,6 +637,9 @@ src/codebud/ the package

src/codebud_sec/ the security layer (authorized security work)
├── engagement.py engagement file: authorized scope
├── scope.py scope guard: every network target read from the
│ command line and checked; registered with the
│ core's safety.add_step_check()
└── evidence.py hash-chained, append-only evidence log,
encrypted at rest (appends need no passphrase)

Expand Down
4 changes: 4 additions & 0 deletions docs/chapters/development.texi
Original file line number Diff line number Diff line change
Expand Up @@ -165,9 +165,13 @@ tests/
│ what would change the host is refused
├── test_shipped_profiles.py profiles of codebud_sec: found by name
│ beside codebud, not imported, not shadowed
├── test_scope.py engagement scope: targets read per tool,
│ third parties, ranges, active refused,
│ passive asked
├── test_skill.py SKILL.md variants (default, Agensi), skill CLI
├── test_step_environment.py steps never see the passphrase or secret
│ settings (any privacy level)
├── test_step_checks.py the core's hook for registered step checks
├── test_steps.py shared step loop: approval policies, the
│ first failure stops the run
├── test_terminal_safety.py control characters in replies, output and
Expand Down
9 changes: 9 additions & 0 deletions docs/chapters/safety-and-sandbox.texi
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,15 @@ command.}), and so is a policy rejection, naming the step and the reason
allow-list}; @pxref{Safety Rejection}). The same reasons are given when a
step is refused right before it runs.

@cindex step checks, registered
An optional layer can add checks of its own: @code{add_step_check()}
registers a function that may refuse a step, and
@code{add_step_notice()} one that may ask for a confirmation. The
engine asks them after its own rules, for every step and again right
before it runs, without knowing what they are; a check that fails with
an error refuses the step. The security layer's engagement scope guard
is registered this way, so the core never imports that layer.

@code{describe_rules()} returns a compact summary of the three policy
lists, which is embedded in the planner prompt so the model can avoid
proposing forbidden commands in the first place.
Expand Down
3 changes: 2 additions & 1 deletion src/Makefile.am
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,8 @@ nobase_python_PYTHON = \
codebud/tools/web_search.py \
codebud_sec/__init__.py \
codebud_sec/engagement.py \
codebud_sec/evidence.py
codebud_sec/evidence.py \
codebud_sec/scope.py

nobase_dist_python_DATA = \
codebud/data/policy/harmless_commands.json \
Expand Down
52 changes: 48 additions & 4 deletions src/codebud/safety.py
Original file line number Diff line number Diff line change
Expand Up @@ -472,6 +472,45 @@ def egress_reason(step: dict[str, Any]) -> str | None:
return None


# Extra checks registered by an optional layer (the security layer's scope
# check, when an engagement is active): the core asks them without
# knowing what they are. A check returns why a step is refused (or None);
# a notice returns why it must be confirmed (or None).
_STEP_CHECKS: list = []
_STEP_NOTICES: list = []


def add_step_check(check) -> None:
_STEP_CHECKS.append(check)


def remove_step_check(check) -> None:
if check in _STEP_CHECKS:
_STEP_CHECKS.remove(check)


def add_step_notice(notice) -> None:
_STEP_NOTICES.append(notice)


def remove_step_notice(notice) -> None:
if notice in _STEP_NOTICES:
_STEP_NOTICES.remove(notice)


def _registered(hooks: list, step: dict[str, Any]) -> str | None:
"""The first reason a registered hook gives for ``step``. A hook that
fails is a refusal: a broken check must never let a step through."""
for hook in list(hooks):
try:
reason = hook(step)
except Exception as exc: # noqa: BLE001 - any failure refuses
return f"a registered step check failed ({type(exc).__name__}: {exc})"
if reason:
return reason
return None


class SafetyEngine:
"""
Validates steps produced by the LLM planner.
Expand Down Expand Up @@ -520,10 +559,12 @@ def refusal_reason(self, step: dict[str, Any]) -> str | None:
if privacy.level() == "locked":
return "steps that need the network are not allowed at privacy level locked"
if tool == "command":
return self._command_refusal(step.get("args", {}).get("cmd", ""))
if tool in ("file_read", "file_write", "patch"):
return _file_refusal(tool, step.get("args", {}))
return None
reason = self._command_refusal(step.get("args", {}).get("cmd", ""))
elif tool in ("file_read", "file_write", "patch"):
reason = _file_refusal(tool, step.get("args", {}))
else:
reason = None
return reason or _registered(_STEP_CHECKS, step)

def is_powerful(self, step: dict[str, Any]) -> bool:
"""Return whether ``step`` is a command on the powerful list.
Expand All @@ -542,6 +583,9 @@ def confirmation_reason(self, step: dict[str, Any]) -> str | None:
Powerful commands run code or change files; egress steps (web
search) send data off the machine.
"""
notice = _registered(_STEP_NOTICES, step)
if notice:
return notice
if self.is_powerful(step):
return POWERFUL_REASON
# At privacy level "trusted" egress runs without asking; safety
Expand Down
18 changes: 18 additions & 0 deletions src/codebud_sec/engagement.py
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,24 @@ def covers(self, target: str) -> bool:
return False
return any(t.matches(target) for t in self.in_scope)

def covers_network(self, network) -> bool:
"""Whether every address of ``network`` (an ipaddress network) is in
scope: inside one in-scope network and overlapping no excluded
network or address. Names never cover addresses."""
inside = any(
t.network is not None
and t.network.version == network.version
and network.subnet_of(t.network)
for t in self.in_scope
)
excluded = any(
t.network is not None
and t.network.version == network.version
and network.overlaps(t.network)
for t in self.out_of_scope
)
return inside and not excluded

def out_of_scope_reason(self, target: str, when: datetime | None = None) -> str | None:
"""Why acting on ``target`` is outside the engagement, or None.

Expand Down
Loading
Loading