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 config/policy.docker.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,9 @@ procedural_guard:
- "find_delete" # find / ... -delete
- "recursive_force_delete" # rm -rf on a non-root path (risk 55, not hard-deny)
- "remote_code_execution" # curl|wget piped into a shell (risk 70)
- "powershell_destruction" # Remove-Item -Recurse -Force C:\, Format-Volume,
# diskpart clean, iwr|iex. Every rule above reads
# POSIX; an agent on Windows writes none of it.
privileged_ops:
- "apt remove"
- "apt purge"
Expand Down
3 changes: 3 additions & 0 deletions config/policy.slim.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,9 @@ procedural_guard:
- "find_delete" # find / ... -delete
- "recursive_force_delete" # rm -rf on a non-root path (risk 55, not hard-deny)
- "remote_code_execution" # curl|wget piped into a shell (risk 70)
- "powershell_destruction" # Remove-Item -Recurse -Force C:\, Format-Volume,
# diskpart clean, iwr|iex. Every rule above reads
# POSIX; an agent on Windows writes none of it.
privileged_ops:
- "apt remove"
- "apt purge"
Expand Down
3 changes: 3 additions & 0 deletions config/policy.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,9 @@ procedural_guard:
- "find_delete" # find / ... -delete
- "recursive_force_delete" # rm -rf on a non-root path (risk 55, not hard-deny)
- "remote_code_execution" # curl|wget piped into a shell (risk 70)
- "powershell_destruction" # Remove-Item -Recurse -Force C:\, Format-Volume,
# diskpart clean, iwr|iex. Every rule above reads
# POSIX; an agent on Windows writes none of it.
privileged_ops:
- "apt remove"
- "apt purge"
Expand Down
109 changes: 109 additions & 0 deletions src/aetherya/procedural_guard.py
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,113 @@ def _detect_recursive_force_delete(text: str) -> bool:
return False


# ---------------------------------------------------------------------------
# PowerShell
#
# Every rule above reads POSIX shell. An agent driving a Windows host writes
# `Remove-Item -Recurse -Force C:\` for what `rm -rf /` means, and that scored
# 0 — the guard did not recognise a single destructive PowerShell form.
#
# Three properties make this harder than the POSIX side:
#
# - Case is insignificant. The shared normalizer lowercases, so patterns are
# written lowercase and that is enough.
# - Cmdlets have aliases, and `rm`, `del` and `rd` are among them, so the
# same token means different things in each shell.
# - Parameters abbreviate to any unambiguous prefix: `-Recurse` is also
# `-recurse`, `-recurs`, `-rec`, `-r`. Matching the full word alone misses
# what an agent actually writes.
#
# And the normalizer strips backslashes before non-space, so `C:\Users\rober`
# arrives as `c:usersrober`. Patterns must not depend on path separators.
# ---------------------------------------------------------------------------

# Remove-Item and its aliases. `rm` and `del` overlap with POSIX tokens, which
# is harmless: the flag shapes below are PowerShell-specific.
_PS_DELETE_CMDLET = r"(?:remove-item|ri\b|rd\b|rmdir\b|del\b|erase\b|rm\b)"

# `-Recurse` / `-Force` accepting any unambiguous prefix, as PowerShell does.
_PS_RECURSE = r"-r(?:e(?:c(?:u(?:r(?:s(?:e)?)?)?)?)?)?\b"
_PS_FORCE = r"-f(?:o(?:r(?:c(?:e)?)?)?)?\b"

# Windows roots as they survive normalization. A drive root keeps its trailing
# backslash only when a space follows it, so both spellings are accepted, and
# the environment variables are matched by name because their expansion is not
# visible here.
_PS_ROOT_TARGET = (
r"(?:[a-z]:\\?(?=\s|$)"
r"|\$env:(?:userprofile|systemroot|systemdrive|windir|programfiles|appdata|homepath)"
r"|%(?:userprofile|systemroot|systemdrive|windir|appdata)%"
r"|\$home\b)"
)


def _detect_powershell_root_delete(text: str) -> bool:
"""
Recursive, forced delete of a Windows root or user profile.

Both flags are required, in any order and any abbreviation, and the target
must be root-like — the same shape as the POSIX rule. Without the target
condition this would fire on `Remove-Item -Recurse -Force .\\build`, which
is ordinary cleanup and the most common thing an agent does.
"""
for match in re.finditer(_PS_DELETE_CMDLET, text):
# One command only: a later pipeline stage is a different invocation.
segment = re.split(r"[|;&\n]", text[match.end() :], maxsplit=1)[0]
if not re.search(_PS_RECURSE, segment):
continue
if not re.search(_PS_FORCE, segment):
continue
if re.search(_PS_ROOT_TARGET, segment):
return True
return False


_POWERSHELL_CRITICAL: tuple[ProceduralRule, ...] = (
_critical("powershell_root_delete", detector=_detect_powershell_root_delete),
_critical(
# Destroys a volume as thoroughly as mkfs. `-Force` is not required:
# these prompt interactively, and an agent supplies confirmation.
"powershell_volume_format",
pattern=r"\b(?:format-volume|clear-disk|initialize-disk|reset-physicaldisk)\b",
),
_critical(
# cmd.exe rather than PowerShell, reachable from the same exec tool.
"windows_format_command",
pattern=r"\bformat\s+[a-z]:\s",
),
_critical(
"windows_partition_wipe",
pattern=r"\bdiskpart\b[^|;&]{0,80}?\bclean\b",
),
)

_POWERSHELL_ELEVATED: tuple[ProceduralRule, ...] = (
ProceduralRule(
tag="remote_code_execution",
risk_score=70,
confidence=0.9,
reason="remote script piped into a shell",
extra_tags=("system_change",),
# `iwr https://x/s.ps1 | iex` is the PowerShell idiom for curl | sh, and
# is what every Windows install one-liner looks like.
pattern=re.compile(
r"\b(?:iwr|irm|invoke-webrequest|invoke-restmethod|curl|wget)\b"
r"[^|;&]{0,200}\|\s*(?:iex\b|invoke-expression\b)"
),
),
ProceduralRule(
tag="remote_code_execution",
risk_score=70,
confidence=0.9,
reason="remote script downloaded and evaluated",
extra_tags=("system_change",),
# The single-expression form, which has no pipe to anchor on.
pattern=re.compile(r"\biex\b[^|;&]{0,40}?\(\s*(?:new-object\s+net\.webclient|iwr|irm)"),
),
)


# ---------------------------------------------------------------------------
# Rule families
#
Expand Down Expand Up @@ -291,6 +398,7 @@ def _detect_recursive_force_delete(text: str) -> bool:
pattern=r"\bfind\s+/\s[^|;&]{0,80}?-delete\b",
),
),
"powershell_destruction": _POWERSHELL_CRITICAL,
}

_ELEVATED_FAMILIES: dict[str, tuple[ProceduralRule, ...]] = {
Expand All @@ -313,6 +421,7 @@ def _detect_recursive_force_delete(text: str) -> bool:
extra_tags=("system_change",),
pattern=re.compile(r"\b(?:curl|wget)\b[^|;&]{0,200}\|\s*(?:sudo\s+)?[a-z]{0,3}sh\b"),
),
*_POWERSHELL_ELEVATED,
),
}

Expand Down
44 changes: 42 additions & 2 deletions src/aetherya/rollout_report.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,11 @@ class RolloutReport:
chain_errors: int = 0
chain_error_detail: str | None = None
hard_deny_events: list[dict[str, Any]] = field(default_factory=list)
# What vocabulary the agent actually speaks: tool -> count, and the
# operations and parameter names seen with it. This is the phase-1
# deliverable that no synthetic corpus can produce — you cannot know which
# tools your runtime calls until you watch it work.
vocabulary: dict[str, dict[str, Any]] = field(default_factory=dict)
criteria: list[Criterion] = field(default_factory=list)

@property
Expand Down Expand Up @@ -89,6 +94,7 @@ def to_dict(self) -> dict[str, Any]:
"detail": self.chain_error_detail,
},
"hard_deny_events": self.hard_deny_events,
"vocabulary": self.vocabulary,
"criteria": [c.to_dict() for c in self.criteria],
}

Expand Down Expand Up @@ -147,6 +153,7 @@ def build_report(
tools: dict[str, set[str]] = {}
fingerprints: set[str] = set()
timestamps: list[datetime] = []
vocabulary: dict[str, dict[str, Any]] = {}

for event in events:
decision = event.get("decision") or {}
Expand All @@ -155,10 +162,22 @@ def build_report(
states[state] += 1

actors.setdefault(state, set()).add(str(event.get("actor", "unknown")))
action_ctx = context.get("action") if isinstance(context.get("action"), dict) else {}
tool = action_ctx.get("tool") if isinstance(action_ctx, dict) else None
raw_action = context.get("action")
action_ctx: dict[str, Any] = raw_action if isinstance(raw_action, dict) else {}
tool = action_ctx.get("tool")
tools.setdefault(state, set()).add(str(tool) if tool else "-")

if tool:
entry = vocabulary.setdefault(
str(tool), {"count": 0, "states": {}, "operations": set(), "parameters": set()}
)
entry["count"] += 1
entry["states"][state] = entry["states"].get(state, 0) + 1
if action_ctx.get("operation"):
entry["operations"].add(str(action_ctx["operation"]))
for name in action_ctx.get("parameter_names") or []:
entry["parameters"].add(str(name))

if context.get("intent_escalation"):
report.intent_escalations += 1
constitution = context.get("constitution")
Expand Down Expand Up @@ -195,6 +214,16 @@ def build_report(
report.actors_by_state = {k: sorted(v) for k, v in sorted(actors.items())}
report.tools_by_state = {k: sorted(v) for k, v in sorted(tools.items())}
report.policy_fingerprints = sorted(fingerprints)
report.vocabulary = {
tool: {
"count": entry["count"],
"states": dict(sorted(entry["states"].items())),
"operations": sorted(entry["operations"]),
"parameters": sorted(entry["parameters"]),
}
# Most frequent first: that is the order to map them in.
for tool, entry in sorted(vocabulary.items(), key=lambda kv: -kv[1]["count"])
}
report.would_block_next_phase = sum(
states[s] for s in (next_phase.blocks | next_phase.confirms) if s in states
)
Expand Down Expand Up @@ -348,6 +377,17 @@ def _print_text(report: RolloutReport) -> None:
print(f" policy fingerprints {len(report.policy_fingerprints)}")
print(f" audit chain {'intact' if report.chain_valid else 'INVALID'}")

if report.vocabulary:
# The phase-1 deliverable. Stop when this stops growing: a thousand
# varied decisions map a runtime better than ten thousand repeats of
# the same call.
print("\n vocabulary observed")
for tool, entry in report.vocabulary.items():
states = " ".join(f"{s}:{n}" for s, n in entry["states"].items())
print(f" {tool:<18} {entry['count']:>4} {states}")
if entry["parameters"]:
print(f" {'':<18} params: {', '.join(entry['parameters'][:8])}")

if report.hard_deny_events:
print(f"\n hard_deny events to review ({len(report.hard_deny_events)} shown)")
for event in report.hard_deny_events[:10]:
Expand Down
4 changes: 2 additions & 2 deletions tests/fixtures/security_baseline/v1/stress_baseline.json
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,10 @@
"attacks_guard_hits": 12,
"attacks_total": 12,
"benign_allow_rate": 1.0,
"benign_allowed": 12,
"benign_allowed": 17,
"benign_guard_hits": 0,
"benign_prompt_safety_violations": 0,
"benign_total": 12
"benign_total": 17
}
},
"version": "v1"
Expand Down
Loading
Loading