diff --git a/README.md b/README.md index 6b97100..aa081f0 100644 --- a/README.md +++ b/README.md @@ -58,10 +58,13 @@ agsec install claude-code # activate the firewall Done. Every tool call is now checked against your policies. Out of the box, the following are blocked: - `rm`, `rm -rf`, `rmdir` — destructive deletes -- Writes to `.env`, `.env.*`, secrets files -- `git push --force` — force pushes -- `DROP TABLE`, `DELETE FROM` without WHERE — destructive SQL -- Reads of `~/.ssh`, `~/.aws`, `~/.gnupg` — credential directories +- Reads and writes to `.env`, credentials, SSH keys, cloud credentials +- `git push --force`, `git reset --hard` — destructive git +- `DROP TABLE`, `TRUNCATE`, `ALTER DROP` — DDL commands +- `DELETE FROM`, `UPDATE SET`, `INSERT INTO` — DML commands +- `sqlite3 audit.db`, `psql audit.db` — audit database tampering +- `chmod 777`, `mkfs`, `dd`, `shred` — destructive filesystem ops +- Direct push to `main`, `master`, `production` branches --- @@ -73,7 +76,7 @@ agsec audit --stats # see what would have been blocked agsec enforce # start blocking when ready ``` -Observe mode gives you a full audit trail of every action your agent attempted — with zero disruption to your workflow. See the blast radius before you enforce it. +Observe mode gives you a full audit trail of every action your agent attempted — with zero disruption to your workflow. See the blast radius before you enforce it. Every action is logged with its actual outcome, so `agsec analyze` accurately shows what got through vs what would have been blocked. --- @@ -110,7 +113,7 @@ statements: actions: ["bash.execute"] ``` -Three effects: `allow`, `deny`, `review` (human-in-the-loop pause). Deny always wins — same evaluation logic as AWS IAM. Layered policy evaluation (project + agent layers) where each layer is a gate. Supports 14 condition operators: `==`, `!=`, `>`, `<`, `>=`, `<=`, `in`, `not_in`, `contains`, `starts_with`, `ends_with`, `regex`, `exists`, `not_exists`. +Three effects: `allow`, `deny`, `review` (human-in-the-loop pause). Deny always wins — same evaluation logic as AWS IAM. Layered policy evaluation (project + agent layers) where each layer is a gate. 21 built-in threat patterns for blast radius analysis. Supports 14 condition operators: `==`, `!=`, `>`, `<`, `>=`, `<=`, `in`, `not_in`, `contains`, `starts_with`, `ends_with`, `regex`, `exists`, `not_exists`. --- @@ -182,6 +185,8 @@ agsec validate # check for errors agsec audit [--stats] # view action log agsec analyze [--hours N] # threat analysis with blast radius +agsec analyze --all # full activity report (every action) +agsec status # firewall status at a glance agsec observe # switch to observe mode agsec enforce # switch to enforce mode diff --git a/agsec/__init__.py b/agsec/__init__.py index 95bbd43..5979d2f 100644 --- a/agsec/__init__.py +++ b/agsec/__init__.py @@ -1,3 +1,5 @@ +__version__ = "0.2.2" + from .audit import AuditStore from .control import ControlLayer from .guard import guard diff --git a/agsec/audit/store.py b/agsec/audit/store.py index 805d269..b65a075 100644 --- a/agsec/audit/store.py +++ b/agsec/audit/store.py @@ -34,6 +34,7 @@ def __init__(self, db_path: Optional[str] = None): self.conn.row_factory = sqlite3.Row self._init_db() + self._auto_prune() def __enter__(self): return self @@ -60,15 +61,25 @@ def _init_db(self) -> None: policy_status TEXT NOT NULL, policy_reason TEXT, context TEXT, - error TEXT + error TEXT, + outcome TEXT ) """) + self.conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_executions_timestamp + ON executions(timestamp) + """) + # Migrate existing databases that don't have the outcome column + try: + self.conn.execute("SELECT outcome FROM executions LIMIT 1") + except sqlite3.OperationalError: + self.conn.execute("ALTER TABLE executions ADD COLUMN outcome TEXT") self.conn.commit() - def log_execution(self, execution: ActionExecutionResult, context: Optional[Dict[str, Any]] = None, error: Optional[str] = None) -> None: + def log_execution(self, execution: ActionExecutionResult, context: Optional[Dict[str, Any]] = None, error: Optional[str] = None, outcome: Optional[str] = None) -> None: self.conn.execute(""" - INSERT INTO executions (timestamp, action, params, result, policy_status, policy_reason, context, error) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) + INSERT INTO executions (timestamp, action, params, result, policy_status, policy_reason, context, error, outcome) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( datetime.utcnow().isoformat(), execution.action, @@ -77,7 +88,8 @@ def log_execution(self, execution: ActionExecutionResult, context: Optional[Dict execution.policy.status.value, execution.policy.reason, json.dumps(context, default=str) if context else None, - error + error, + outcome, )) self.conn.commit() @@ -127,6 +139,30 @@ def get_executions_since( rows = self.conn.execute(query, params).fetchall() return [dict(row) for row in rows] + def prune(self, days: int = 7) -> int: + """Delete audit records older than N days. Returns count deleted.""" + cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat() + cursor = self.conn.execute("DELETE FROM executions WHERE timestamp < ?", (cutoff,)) + self.conn.commit() + return cursor.rowcount + + def clear(self) -> int: + """Delete ALL audit records. Returns count deleted.""" + cursor = self.conn.execute("DELETE FROM executions") + self.conn.commit() + return cursor.rowcount + + def _auto_prune(self) -> None: + """Auto-prune if AGSEC_AUDIT_RETENTION_DAYS is set.""" + retention = os.environ.get("AGSEC_AUDIT_RETENTION_DAYS") + if retention is not None: + try: + days = int(retention) + if days > 0: + self.prune(days=days) + except ValueError: + pass + def export_to_json(self, file_path: str) -> None: executions = self.get_executions(limit=10000) # Write with restricted permissions diff --git a/agsec/cli/commands/analyze.py b/agsec/cli/commands/analyze.py index 3ed76ca..ed9e367 100644 --- a/agsec/cli/commands/analyze.py +++ b/agsec/cli/commands/analyze.py @@ -8,12 +8,35 @@ from ...audit import AuditStore from ...threat import Severity, ThreatClassifier, group_findings from ..config import get_audit_db_path, load_mode +from ..output import ( + error, heading, info, mode_label, plain, severity_label, + status_allow, status_block, status_review, success, subheading, warn, +) + + +# Human-readable labels for action types: (display_name, param_field_for_value) +ACTION_LABELS = { + "bash.execute": ("Shell Commands", "command"), + "file.edit": ("File Edits", "file_path"), + "file.write": ("File Creation", "file_path"), + "file.read": ("File Reads", "file_path"), + "web.fetch": ("Web Requests", "url"), + "web.search": ("Web Searches", "query"), + "file.glob": ("File Search", "pattern"), + "file.grep": ("Content Search", "pattern"), + "agent.spawn": ("Sub-Agent Spawns", "prompt"), + "notebook.edit": ("Notebook Edits", "file_path"), +} + +MAX_ITEMS_PER_GROUP = 5 def register(subparsers): p = subparsers.add_parser("analyze", help="Threat analysis of observed agent behavior") p.add_argument("--hours", type=float, help="Analyze last N hours") p.add_argument("--days", type=float, help="Analyze last N days") + p.add_argument("--all", dest="show_all", action="store_true", + help="Show all actions grouped by type (full activity report)") p.add_argument("--json", dest="as_json", action="store_true", help="Output as JSON") p.set_defaults(func=run) @@ -22,29 +45,35 @@ def run(args): try: audit = AuditStore(get_audit_db_path()) except Exception as e: - print(f"Error opening audit database: {e}", file=sys.stderr) + error(f"Cannot open audit database: {e}") + info("Fix: Check permissions on ~/.agsec/audit.db or set AGSEC_AUDIT_DB") sys.exit(1) executions = audit.get_executions_since(hours=args.hours, days=args.days) if not executions: - print("No audit records found. Run your agent in observe mode first.") + info("No audit records found. Run your agent in observe mode first.") return - classifier = ThreatClassifier() - report = classifier.classify(executions) mode = load_mode() - if args.as_json: - _render_json(report, mode, args) + if args.show_all: + groups = _build_activity_groups(executions) + if args.as_json: + _render_all_json(groups, executions, mode, args) + else: + _render_all_human(groups, executions, mode, args) else: - _render_human(report, mode, args) + classifier = ThreatClassifier() + report = classifier.classify(executions) + if args.as_json: + _render_json(report, mode, args) + else: + _render_human(report, mode, args) def _render_human(report, mode, args): - # Header - print("agsec threat analysis") - print("=" * 21) + heading("agsec threat analysis") # Time window if args.hours: @@ -53,17 +82,20 @@ def _render_human(report, mode, args): window = f"Last {args.days}d" else: window = "All time" - print(f"Time window: {window} ({report.total_executions} actions)") - print(f"Mode: {mode}") - print() + plain(f"Time window: {window} ({report.total_executions} actions)") + plain(f"Mode: {mode_label(mode)}") + if mode == "observe": + warn("Observe mode: scores reflect what WOULD have been blocked.") + info("Switch to enforce mode with: agsec enforce") + plain("") # Blast radius bar_filled = int(report.blast_radius) bar_empty = 10 - bar_filled bar = "\u2588" * bar_filled + "\u2591" * bar_empty - print(f"Blast Radius: {report.blast_radius} / 10.0 \u2014 {report.blast_radius_label}") - print(f" {bar}") - print() + plain(f"Blast Radius: {report.blast_radius} / 10.0 \u2014 {report.blast_radius_label}") + plain(f" {bar}") + plain("") # Threats (allowed through) if report.threats: @@ -71,38 +103,38 @@ def _render_human(report, mode, args): current_severity = None for group in grouped: - sev = group["severity"].upper() - if sev != current_severity: - current_severity = sev - sev_count = report.severity_counts.get(group["severity"], 0) - print(f"{sev} ({sev_count} unblocked)") + sev = group["severity"] + if sev.upper() != current_severity: + current_severity = sev.upper() + sev_count = report.severity_counts.get(sev, 0) + subheading(f"{severity_label(sev)} ({sev_count} unblocked)") - print(f" {group['name']} ({group['count']}x)") + plain(f" {group['name']} ({group['count']}x)") if group["examples"]: examples_str = ", ".join(group["examples"][:3]) if len(examples_str) > 100: examples_str = examples_str[:97] + "..." - print(f" \u2192 {examples_str}") - print(f" Impact: {group['consequence']}") - print() + info(f" \u2192 {examples_str}") + info(f" Impact: {group['consequence']}") + plain("") else: - print("No unblocked threats detected.") - print() + success("No unblocked threats detected.") + plain("") # Caught by policy if report.blocked: total_blocked = len(report.blocked) - print(f"Caught by policy ({total_blocked} actions blocked)") + success(f"Caught by policy ({total_blocked} actions blocked)") blocked_grouped = group_findings(report.blocked) summaries = [f"{g['count']}x {g['name'].lower()}" for g in blocked_grouped] - print(f" {', '.join(summaries)}") - print() + info(f" {', '.join(summaries)}") + plain("") # Recommendations if report.recommendations: - print("Recommendations") + subheading("Recommendations") for i, rec in enumerate(report.recommendations, 1): - print(f" {i}. {rec}") + plain(f" {i}. {rec}") def _render_json(report, mode, args): @@ -128,3 +160,180 @@ def _render_json(report, mode, args): output["time_window"] = "all" print(json.dumps(output, indent=2)) + + +# --------------------------------------------------------------------------- +# --all: Full activity report +# --------------------------------------------------------------------------- + + +def _get_time_window(args): + if args.hours: + return f"Last {args.hours}h" + elif args.days: + return f"Last {args.days}d" + return "All time" + + +def _build_activity_groups(executions): + """Group all executions by action type with human-readable labels.""" + groups = {} + + for row in executions: + action = row.get("action", "") + status = row.get("policy_status", "allow") + + # Skip internal tools — not interesting to users + if action.startswith("internal."): + continue + + raw_params = row.get("params", "{}") + if isinstance(raw_params, str): + try: + params = json.loads(raw_params) + except (json.JSONDecodeError, TypeError): + params = {} + elif isinstance(raw_params, dict): + params = raw_params + else: + params = {} + + # Resolve label and value field + if action in ACTION_LABELS: + label, value_field = ACTION_LABELS[action] + elif action.startswith("mcp."): + label = f"MCP: {action[4:]}" + value_field = None + else: + label = action + value_field = None + + if action not in groups: + groups[action] = { + "label": label, + "items": [], + "counts": {"allow": 0, "block": 0, "review": 0}, + } + + groups[action]["counts"][status] = groups[action]["counts"].get(status, 0) + 1 + + # Extract display value + if value_field and value_field in params: + value = str(params[value_field]) + elif params: + value = str(list(params.values())[0]) + else: + value = "(no params)" + + # Truncate long values + if len(value) > 100: + value = value[:97] + "..." + + groups[action]["items"].append({"value": value, "status": status}) + + return groups + + +def _render_all_human(groups, executions, mode, args): + heading("agsec activity report") + + window = _get_time_window(args) + total = len([e for e in executions if not e.get("action", "").startswith("internal.")]) + plain(f"Time window: {window} ({total} actions) | Mode: {mode_label(mode)}") + plain("") + + is_observe = mode == "observe" + + # Sort: groups with blocks/reviews first, then by total count descending + def sort_key(item): + g = item[1] + has_issues = g["counts"].get("block", 0) + g["counts"].get("review", 0) + total_count = sum(g["counts"].values()) + return (-has_issues, -total_count) + + sorted_groups = sorted(groups.items(), key=sort_key) + + for action, group in sorted_groups: + label = group["label"] + counts = group["counts"] + total_count = sum(counts.values()) + + # Build count summary + parts = [] + if counts.get("allow", 0): + parts.append(f"{counts['allow']} allow") + if counts.get("block", 0): + block_word = "would-block" if is_observe else "block" + parts.append(f"{counts['block']} {block_word}") + if counts.get("review", 0): + parts.append(f"{counts['review']} review") + + summary = ", ".join(parts) + subheading(f"{label} ({total_count} actions \u2014 {summary})") + + # Sort items: block first, then review, then allow + status_order = {"block": 0, "review": 1, "allow": 2} + sorted_items = sorted(group["items"], key=lambda x: status_order.get(x["status"], 3)) + + # Deduplicate values while preserving order and worst status + seen = {} + for item in sorted_items: + val = item["value"] + if val not in seen: + seen[val] = item + # Keep the worst status (block > review > allow) + elif status_order.get(item["status"], 3) < status_order.get(seen[val]["status"], 3): + seen[val] = item + + deduped = list(seen.values()) + + shown = 0 + for item in deduped: + if shown >= MAX_ITEMS_PER_GROUP: + break + value = item["value"] + status = item["status"] + + if status == "block": + status_block(f" {value}") + elif status == "review": + status_review(f" {value}") + else: + status_allow(f" {value}") + shown += 1 + + remaining = len(deduped) - shown + if remaining > 0: + info(f" ... {remaining} more") + + plain("") + + +def _render_all_json(groups, executions, mode, args): + total = len([e for e in executions if not e.get("action", "").startswith("internal.")]) + + output = { + "mode": mode, + "total_actions": total, + "groups": {}, + } + + if args.hours: + output["time_window"] = f"{args.hours}h" + elif args.days: + output["time_window"] = f"{args.days}d" + else: + output["time_window"] = "all" + + for action, group in groups.items(): + counts = group["counts"] + output["groups"][action] = { + "label": group["label"], + "total": sum(counts.values()), + "allow": counts.get("allow", 0), + "block": counts.get("block", 0), + "review": counts.get("review", 0), + "items": group["items"], + } + + print(json.dumps(output, indent=2)) diff --git a/agsec/cli/commands/audit.py b/agsec/cli/commands/audit.py index 1f77196..bf765ba 100644 --- a/agsec/cli/commands/audit.py +++ b/agsec/cli/commands/audit.py @@ -7,6 +7,7 @@ from ...audit import AuditStore from ..config import get_audit_db_path, load_mode +from ..output import error, heading, info, mode_label, plain, status_allow, status_block, status_review, success, table def register(subparsers): @@ -15,6 +16,10 @@ def register(subparsers): p.add_argument("--action", help="Filter by action name") p.add_argument("--limit", type=int, default=20, help="Number of records (default: 20)") p.add_argument("--json", dest="as_json", action="store_true", help="Output as JSON") + p.add_argument("--prune", type=int, nargs="?", const=7, metavar="DAYS", + help="Delete records older than N days (default: 7)") + p.add_argument("--clear", action="store_true", + help="Delete ALL audit records") p.set_defaults(func=run) @@ -22,9 +27,20 @@ def run(args): try: audit = AuditStore(get_audit_db_path()) except Exception as e: - print(f"Error opening audit database: {e}", file=sys.stderr) + error(f"Cannot open audit database: {e}") + info("Fix: Check permissions on ~/.agsec/audit.db or set AGSEC_AUDIT_DB") sys.exit(1) + if args.clear: + count = audit.clear() + success(f"Cleared {count} audit records.") + return + + if args.prune is not None: + count = audit.prune(days=args.prune) + success(f"Pruned {count} records older than {args.prune} days.") + return + if args.stats: stats = audit.get_execution_stats() mode = load_mode() @@ -32,32 +48,40 @@ def run(args): stats["mode"] = mode print(json.dumps(stats, indent=2)) else: - mode_label = "OBSERVE" if mode == "observe" else "ENFORCE" - print(f"Audit Statistics ({mode_label} mode)") - print(f" Total: {stats['total_executions']}") - print(f" Allowed: {stats['allowed']}") + heading(f"Audit Statistics ({mode_label(mode)} mode)") + plain(f" Total: {stats['total_executions']}") + plain(f" Allowed: {stats['allowed']}") if mode == "observe": - print(f" Would block: {stats['blocked']}") - print(f" Would review: {stats['reviewed']}") + plain(f" Would block: {stats['blocked']}") + plain(f" Would review: {stats['reviewed']}") else: - print(f" Blocked: {stats['blocked']}") - print(f" Reviewed: {stats['reviewed']}") - print(f" Errors: {stats['errors']}") + plain(f" Blocked: {stats['blocked']}") + plain(f" Reviewed: {stats['reviewed']}") + plain(f" Errors: {stats['errors']}") return executions = audit.get_executions(action=args.action, limit=args.limit) if not executions: - print("No audit records found.") + info("No audit records found.") return if args.as_json: print(json.dumps(executions, indent=2, default=str)) else: for ex in executions: - status = ex["policy_status"].upper() - icon = {"allow": "+", "block": "X", "review": "?"} - marker = icon.get(ex["policy_status"], " ") - print(f" [{marker}] {ex['timestamp']} {status:6s} {ex['action']}") - if ex.get("policy_reason"): - print(f" {ex['policy_reason']}") + status = ex["policy_status"] + ts = ex["timestamp"] + action = ex["action"] + reason = ex.get("policy_reason", "") + + line = f"{ts} {status.upper():6s} {action}" + if status == "allow": + status_allow(line) + elif status == "block": + status_block(line) + else: + status_review(line) + + if reason: + info(f" {reason}") diff --git a/agsec/cli/commands/check.py b/agsec/cli/commands/check.py index 34e6474..c9b9aed 100644 --- a/agsec/cli/commands/check.py +++ b/agsec/cli/commands/check.py @@ -123,11 +123,23 @@ def run(args): mode = load_mode() context["agsec_mode"] = mode + # Determine actual outcome: what really happened to this action + if mode == "halt": + outcome = "blocked" + elif mode == "observe": + outcome = "allowed" # observe lets everything through + elif result.status == PolicyStatus.BLOCK: + outcome = "blocked" + elif result.status == PolicyStatus.REVIEW: + outcome = "review" + else: + outcome = "allowed" + # Audit log (never fail the check due to audit) try: - audit = AuditStore(get_audit_db_path()) - exec_result = ActionExecutionResult(action=action, params=params, result=None, policy=result) - audit.log_execution(exec_result, context) + with AuditStore(get_audit_db_path()) as audit: + exec_result = ActionExecutionResult(action=action, params=params, result=None, policy=result) + audit.log_execution(exec_result, context, outcome=outcome) except Exception: pass diff --git a/agsec/cli/commands/init.py b/agsec/cli/commands/init.py index fb96495..c7ea4ec 100644 --- a/agsec/cli/commands/init.py +++ b/agsec/cli/commands/init.py @@ -6,6 +6,7 @@ import shutil from ..config import get_templates_dir, set_mode +from ..output import error, info, success, warn def register(subparsers): @@ -20,13 +21,13 @@ def run(args): target = os.path.join(os.getcwd(), args.dir) if os.path.exists(target): - print(f"Directory already exists: {target}") - print("Use 'agsec validate' to check existing policies.") + warn(f"Directory already exists: {target}") + info("Use 'agsec validate' to check existing policies.") return templates = get_templates_dir() if not os.path.isdir(templates): - print("Error: Template policies not found. Reinstall agsec.") + error("Template policies not found. Reinstall agsec.") return shutil.copytree(templates, target) @@ -35,19 +36,18 @@ def run(args): mode = "observe" if args.observe else "enforce" config_path = set_mode(mode) - mode_label = "OBSERVE" if args.observe else "ENFORCE" + mode_str = "OBSERVE" if args.observe else "ENFORCE" files = sorted(os.listdir(target)) - print(f"Created {target}/ with {len(files)} policy files ({mode_label} mode)") + success(f"Created {target}/ with {len(files)} policy files ({mode_str} mode)") for f in files: - print(f" {f}") - print() + info(f" {f}") if args.observe: - print("Observe mode: all actions are ALLOWED but logged.") - print("Run 'agsec audit --stats' to see what would be blocked.") - print("Run 'agsec enforce' when ready to start blocking.") + warn("Observe mode: all actions are ALLOWED but logged.") + info("Run 'agsec audit --stats' to see what would be blocked.") + info("Run 'agsec enforce' when ready to start blocking.") else: - print("Next steps:") - print(" 1. Edit policies to match your needs") - print(" 2. Run 'agsec validate' to check for errors") - print(" 3. Run 'agsec install claude-code' or 'agsec install codex' to activate") + info("Next steps:") + info(" 1. Edit policies to match your needs") + info(" 2. Run 'agsec validate' to check for errors") + info(" 3. Run 'agsec install claude-code' or 'agsec install codex' to activate") diff --git a/agsec/cli/commands/install.py b/agsec/cli/commands/install.py index caf46ea..cc39e89 100644 --- a/agsec/cli/commands/install.py +++ b/agsec/cli/commands/install.py @@ -8,6 +8,8 @@ import shutil import sys +from ..output import error, info, success, warn + def register(subparsers): p = subparsers.add_parser("install", help="Install agsec hooks for an agent platform") @@ -95,8 +97,8 @@ def _install_claude_code(project_dir: str): ) if already_installed: - print("agsec hook already installed in Claude Code.") - print(f" Config: {settings_path}") + warn("agsec hook already installed in Claude Code.") + info(f" Config: {settings_path}") return pre_tool_hooks.append(new_hook) @@ -105,11 +107,10 @@ def _install_claude_code(project_dir: str): with os.fdopen(fd, "w") as f: json.dump(settings, f, indent=2) - print("agsec hook installed for Claude Code.") - print(f" Config: {settings_path}") - print(f" Hook: {hook_command}") - print() - print("The firewall is now active. Every tool call will be checked against your policies.") + success("agsec hook installed for Claude Code.") + info(f" Config: {settings_path}") + info(f" Hook: {hook_command}") + info("The firewall is now active. Every tool call will be checked against your policies.") def _install_codex(project_dir: str): @@ -137,8 +138,8 @@ def _install_codex(project_dir: str): # Check if already installed for h in existing.get("hooks", []): if "agsec" in h.get("command", ""): - print("agsec hook already installed for Codex.") - print(f" Config: {hooks_path}") + warn("agsec hook already installed for Codex.") + info(f" Config: {hooks_path}") return existing.setdefault("hooks", []).append(hooks_config["hooks"][0]) hooks_config = existing @@ -149,9 +150,9 @@ def _install_codex(project_dir: str): with os.fdopen(fd, "w") as f: json.dump(hooks_config, f, indent=2) - print("agsec hook installed for Codex.") - print(f" Config: {hooks_path}") - print(f" Hook: {hook_command}") + success("agsec hook installed for Codex.") + info(f" Config: {hooks_path}") + info(f" Hook: {hook_command}") # --------------------------------------------------------------------------- @@ -176,14 +177,14 @@ def _uninstall_claude_code(project_dir: str): settings_path = os.path.join(project_dir, ".claude", "settings.json") if not os.path.isfile(settings_path): - print("agsec is not installed for Claude Code (no settings.json found).") + warn("agsec is not installed for Claude Code (no settings.json found).") return with open(settings_path, "r") as f: try: settings = json.load(f) except json.JSONDecodeError: - print("Could not read settings.json.") + error("Could not read settings.json.") return hooks = settings.get("hooks", {}) @@ -196,7 +197,7 @@ def _uninstall_claude_code(project_dir: str): ] if len(filtered) == len(pre_tool_hooks): - print("agsec is not installed for Claude Code.") + warn("agsec is not installed for Claude Code.") return # Clean up empty structures @@ -214,30 +215,30 @@ def _uninstall_claude_code(project_dir: str): with os.fdopen(fd, "w") as f: json.dump(settings, f, indent=2) - print("agsec hook removed from Claude Code.") - print(f" Config: {settings_path}") - print(" Restart Claude Code for changes to take effect.") + success("agsec hook removed from Claude Code.") + info(f" Config: {settings_path}") + info(" Restart Claude Code for changes to take effect.") def _uninstall_codex(project_dir: str): hooks_path = os.path.join(project_dir, ".codex", "hooks.json") if not os.path.isfile(hooks_path): - print("agsec is not installed for Codex (no hooks.json found).") + warn("agsec is not installed for Codex (no hooks.json found).") return with open(hooks_path, "r") as f: try: config = json.load(f) except json.JSONDecodeError: - print("Could not read hooks.json.") + error("Could not read hooks.json.") return hooks_list = config.get("hooks", []) filtered = [h for h in hooks_list if "agsec" not in h.get("command", "")] if len(filtered) == len(hooks_list): - print("agsec is not installed for Codex.") + warn("agsec is not installed for Codex.") return config["hooks"] = filtered @@ -246,8 +247,8 @@ def _uninstall_codex(project_dir: str): with os.fdopen(fd, "w") as f: json.dump(config, f, indent=2) - print("agsec hook removed from Codex.") - print(f" Config: {hooks_path}") + success("agsec hook removed from Codex.") + info(f" Config: {hooks_path}") # --------------------------------------------------------------------------- @@ -281,14 +282,14 @@ def _install_cursor(project_dir: str): shell_hooks = hooks.setdefault("beforeShellExecution", []) if any("agsec" in h.get("command", "") for h in shell_hooks): - print("agsec hook already installed for Cursor.") + warn("agsec hook already installed for Cursor.") return shell_hooks.append(hook_entry) _write_json(hooks_path, config) - print("agsec hook installed for Cursor.") - print(f" Config: {hooks_path}") + success("agsec hook installed for Cursor.") + info(f" Config: {hooks_path}") def _uninstall_cursor(project_dir: str): @@ -330,14 +331,14 @@ def _install_windsurf(project_dir: str): pre_hooks = hooks.setdefault("PreToolUse", []) if any("agsec" in str(h) for h in pre_hooks): - print("agsec hook already installed for Windsurf.") + warn("agsec hook already installed for Windsurf.") return pre_hooks.append(new_hook) _write_json(settings_path, settings) - print("agsec hook installed for Windsurf.") - print(f" Config: {settings_path}") + success("agsec hook installed for Windsurf.") + info(f" Config: {settings_path}") def _uninstall_windsurf(project_dir: str): @@ -363,7 +364,7 @@ def _install_cline(project_dir: str): cmd += f" --policy-dir {shlex.quote(policy_dir)}" if os.path.isfile(script_path): - print("agsec hook already installed for Cline.") + warn("agsec hook already installed for Cline.") return script = f"""#!/bin/bash @@ -374,20 +375,20 @@ def _install_cline(project_dir: str): with os.fdopen(fd, "w") as f: f.write(script) - print("agsec hook installed for Cline.") - print(f" Script: {script_path}") + success("agsec hook installed for Cline.") + info(f" Script: {script_path}") def _uninstall_cline(project_dir: str): script_path = os.path.join(project_dir, ".clinerules", "hooks", "agsec-check.sh") if not os.path.isfile(script_path): - print("agsec is not installed for Cline.") + warn("agsec is not installed for Cline.") return os.unlink(script_path) - print("agsec hook removed from Cline.") - print(f" Removed: {script_path}") + success("agsec hook removed from Cline.") + info(f" Removed: {script_path}") # --------------------------------------------------------------------------- @@ -438,7 +439,7 @@ def _install_copilot(project_dir: str): pre_hooks = hooks.setdefault("preToolUse", []) if any("agsec" in str(h.get("bash", "")) for h in pre_hooks): - print("agsec hook already installed for GitHub Copilot.") + warn("agsec hook already installed for GitHub Copilot.") return pre_hooks.append(hook_entry) @@ -451,12 +452,11 @@ def _install_copilot(project_dir: str): user_hooks_path = os.path.join(user_hooks_dir, "hooks.json") _write_json(user_hooks_path, config) - print("agsec hook installed for GitHub Copilot.") - print(f" Project: {hooks_path} (commit and push for cloud agent)") - print(f" User: {user_hooks_path} (local VS Code)") - print() - print(" Note: 'agsec install claude-code' also covers VS Code Copilot") - print(" since VS Code reads .claude/settings.json hooks.") + success("agsec hook installed for GitHub Copilot.") + info(f" Project: {hooks_path} (commit and push for cloud agent)") + info(f" User: {user_hooks_path} (local VS Code)") + info(" Note: 'agsec install claude-code' also covers VS Code Copilot") + info(" since VS Code reads .claude/settings.json hooks.") def _uninstall_copilot(project_dir: str): @@ -466,20 +466,20 @@ def _uninstall_copilot(project_dir: str): project_path = os.path.join(project_dir, ".github", "hooks", "hooks.json") if os.path.isfile(project_path): if _remove_agsec_from_hooks_file(project_path): - print(f" Removed from: {project_path}") + info(f" Removed from: {project_path}") removed = True # Remove from ~/.copilot/hooks/ user_path = os.path.join(os.path.expanduser("~"), ".copilot", "hooks", "hooks.json") if os.path.isfile(user_path): if _remove_agsec_from_hooks_file(user_path): - print(f" Removed from: {user_path}") + info(f" Removed from: {user_path}") removed = True if removed: - print("agsec hook removed from GitHub Copilot.") + success("agsec hook removed from GitHub Copilot.") else: - print("agsec is not installed for GitHub Copilot.") + warn("agsec is not installed for GitHub Copilot.") def _remove_agsec_from_hooks_file(path: str) -> bool: @@ -528,40 +528,40 @@ def _write_json(path: str, data: dict): def _uninstall_json_hooks(hooks_path: str, name: str, hook_key: str): """Remove agsec hooks from a JSON hooks file with a list under hook_key.""" if not os.path.isfile(hooks_path): - print(f"agsec is not installed for {name}.") + warn(f"agsec is not installed for {name}.") return with open(hooks_path, "r") as f: try: config = json.load(f) except json.JSONDecodeError: - print(f"Could not read {os.path.basename(hooks_path)}.") + error(f"Could not read {os.path.basename(hooks_path)}.") return hooks_list = config.get("hooks", {}).get(hook_key, []) filtered = [h for h in hooks_list if "agsec" not in h.get("command", "")] if len(filtered) == len(hooks_list): - print(f"agsec is not installed for {name}.") + warn(f"agsec is not installed for {name}.") return config["hooks"][hook_key] = filtered _write_json(hooks_path, config) - print(f"agsec hook removed from {name}.") - print(f" Config: {hooks_path}") + success(f"agsec hook removed from {name}.") + info(f" Config: {hooks_path}") def _uninstall_settings_hooks(settings_path: str, name: str): """Remove agsec hooks from a settings.json with PreToolUse structure.""" if not os.path.isfile(settings_path): - print(f"agsec is not installed for {name}.") + warn(f"agsec is not installed for {name}.") return with open(settings_path, "r") as f: try: settings = json.load(f) except json.JSONDecodeError: - print(f"Could not read {os.path.basename(settings_path)}.") + error(f"Could not read {os.path.basename(settings_path)}.") return hooks = settings.get("hooks", {}) @@ -569,7 +569,7 @@ def _uninstall_settings_hooks(settings_path: str, name: str): filtered = [h for h in pre_hooks if "agsec" not in str(h)] if len(filtered) == len(pre_hooks): - print(f"agsec is not installed for {name}.") + warn(f"agsec is not installed for {name}.") return if filtered: @@ -583,5 +583,5 @@ def _uninstall_settings_hooks(settings_path: str, name: str): settings.pop("hooks", None) _write_json(settings_path, settings) - print(f"agsec hook removed from {name}.") - print(f" Config: {settings_path}") + success(f"agsec hook removed from {name}.") + info(f" Config: {settings_path}") diff --git a/agsec/cli/commands/mode.py b/agsec/cli/commands/mode.py index 558775b..fcdb130 100644 --- a/agsec/cli/commands/mode.py +++ b/agsec/cli/commands/mode.py @@ -3,6 +3,7 @@ from __future__ import annotations from ..config import get_previous_mode, load_mode, set_mode +from ..output import error, info, success, warn def register_observe(subparsers): @@ -18,25 +19,23 @@ def register_enforce(subparsers): def run_observe(args): current = load_mode() if current == "observe": - print("Already in OBSERVE mode.") + warn("Already in OBSERVE mode.") return - config_path = set_mode("observe") - print("Switched to OBSERVE mode.") - print(" All actions are allowed but logged.") - print(" Run 'agsec audit --stats' to see what would be blocked.") - print(" Run 'agsec enforce' when ready to start blocking.") - print(f" Config: {config_path}") + set_mode("observe") + warn("Switched to OBSERVE mode.") + info("All actions are allowed but logged.") + info("Run 'agsec audit --stats' to see what would be blocked.") + info("Run 'agsec enforce' when ready to start blocking.") def run_enforce(args): current = load_mode() if current == "enforce": - print("Already in ENFORCE mode.") + success("Already in ENFORCE mode.") return - config_path = set_mode("enforce") - print("Switched to ENFORCE mode.") - print(" Policies are now enforced. Blocked actions will be denied.") - print(f" Config: {config_path}") + set_mode("enforce") + success("Switched to ENFORCE mode.") + info("Policies are now enforced. Blocked actions will be denied.") def register_halt(subparsers): @@ -52,20 +51,18 @@ def register_resume(subparsers): def run_halt(args): current = load_mode() if current == "halt": - print("Already HALTED. All actions are blocked.") + error("Already HALTED. All actions are blocked.") return - config_path = set_mode("halt", store_previous=True) - print("HALTED. All agent actions are now blocked.") - print(f" Previous mode ({current.upper()}) saved. Run 'agsec resume' to restore.") - print(f" Config: {config_path}") + set_mode("halt", store_previous=True) + error("HALTED. All agent actions are now blocked.") + info(f"Previous mode ({current.upper()}) saved. Run 'agsec resume' to restore.") def run_resume(args): current = load_mode() if current != "halt": - print(f"Not halted. Current mode: {current.upper()}") + info(f"Not halted. Current mode: {current.upper()}") return previous = get_previous_mode() - config_path = set_mode(previous) - print(f"Resumed. Restored to {previous.upper()} mode.") - print(f" Config: {config_path}") + set_mode(previous) + success(f"Resumed. Restored to {previous.upper()} mode.") diff --git a/agsec/cli/commands/policy.py b/agsec/cli/commands/policy.py index bab538e..c2e8598 100644 --- a/agsec/cli/commands/policy.py +++ b/agsec/cli/commands/policy.py @@ -8,6 +8,7 @@ import yaml from ..config import find_policy_dir +from ..output import error, info, plain, status_allow, status_block, status_review, subheading, success # Known actions for interactive picker KNOWN_ACTIONS = { @@ -303,7 +304,7 @@ def run_add(args): try: policy_dir = find_policy_dir() except FileNotFoundError: - print("No policies directory found. Run 'agsec init' first.", file=sys.stderr) + error("No policies directory found. Run 'agsec init' first.") sys.exit(1) # Interactive mode if required args are missing @@ -336,7 +337,7 @@ def run_add(args): key, op, value = _parse_condition(cond_str) conditions[key] = {"op": op, "value": value} except ValueError as e: - print(f"Error: {e}", file=sys.stderr) + error(str(e)) sys.exit(1) # Build statement @@ -364,7 +365,7 @@ def run_add(args): existing = doc.get("statements", []) for s in existing: if s.get("sid") == sid: - print(f"Error: Statement '{sid}' already exists in {os.path.basename(target)}", file=sys.stderr) + error(f"Statement '{sid}' already exists in {os.path.basename(target)}") sys.exit(1) existing.append(statement) @@ -374,23 +375,26 @@ def run_add(args): yaml.dump(doc, f, default_flow_style=False, sort_keys=False) # Summary - print(f"\n Added to {os.path.basename(target)}:") - icon = {"deny": "X", "allow": "+", "review": "?"} - print(f" [{icon.get(effect, ' ')}] {sid} ({effect.upper()})") - print(f" Actions: {', '.join(actions)}") + success(f"Added to {os.path.basename(target)}:") + line = f"{sid} ({effect.upper()}) {', '.join(actions)}" + if effect == "allow": + status_allow(line) + elif effect in ("deny", "block"): + status_block(line) + else: + status_review(line) if conditions: for key, cond in conditions.items(): - print(f" Condition: {key} {cond['op']} {cond.get('value', '')}") + info(f" Condition: {key} {cond['op']} {cond.get('value', '')}") if reason: - print(f" Reason: {reason}") - print() + info(f" Reason: {reason}") def run_remove(args): try: policy_dir = find_policy_dir() except FileNotFoundError: - print("No policies directory found.", file=sys.stderr) + error("No policies directory found.") sys.exit(1) files = sorted( @@ -412,12 +416,12 @@ def run_remove(args): doc["statements"] = statements with open(path, "w") as f: yaml.dump(doc, f, default_flow_style=False, sort_keys=False) - print(f"Removed statement '{args.sid}' from {filename}") + success(f"Removed statement '{args.sid}' from {filename}") found = True break if not found: - print(f"Statement '{args.sid}' not found in any policy file.", file=sys.stderr) + error(f"Statement '{args.sid}' not found in any policy file.") sys.exit(1) @@ -425,7 +429,8 @@ def run_list(args): try: policy_dir = find_policy_dir() except FileNotFoundError: - print("No policies directory found. Run 'agsec init' first.", file=sys.stderr) + error("No policies directory found.") + info("Run 'agsec init' first.") sys.exit(1) files = sorted( @@ -434,7 +439,7 @@ def run_list(args): ) if not files: - print("No policy files found.") + info("No policy files found.") return total = 0 @@ -450,29 +455,32 @@ def run_list(args): if not statements and not rules: continue - print(f"\n{filename}" + (f" (default: {default})" if default else "")) - print("-" * 60) + subheading(f"{filename}" + (f" (default: {default})" if default else "")) for s in statements: sid = s.get("sid", "(no sid)") effect = s.get("effect", "?").upper() actions = s.get("actions", ["*"]) reason = s.get("reason", "") - icon = {"ALLOW": "+", "DENY": "X", "BLOCK": "X", "REVIEW": "?"} - marker = icon.get(effect, " ") - print(f" [{marker}] {sid:30s} {effect:6s} {', '.join(actions)}") + line = f"{sid:30s} {effect:6s} {', '.join(actions)}" + if effect in ("DENY", "BLOCK"): + status_block(line) + elif effect == "ALLOW": + status_allow(line) + else: + status_review(line) if reason: - print(f" {reason}") + info(f" {reason}") total += 1 for r in rules: action = r.get("action", "*") status = r.get("status", "?").upper() reason = r.get("reason", "") - print(f" [ ] (legacy rule) {status:6s} {action}") + plain(f" [ ] (legacy rule) {status:6s} {action}") if reason: - print(f" {reason}") + info(f" {reason}") total += 1 - print(f"\n{total} statement(s) across {len(files)} file(s)") + plain(f"\n{total} statement(s) across {len(files)} file(s)") diff --git a/agsec/cli/commands/status.py b/agsec/cli/commands/status.py new file mode 100644 index 0000000..b60750f --- /dev/null +++ b/agsec/cli/commands/status.py @@ -0,0 +1,168 @@ +"""agsec status — show current firewall state at a glance.""" + +from __future__ import annotations + +import glob +import json +import os +import sys + +from ... import __version__ +from ...audit import AuditStore +from ..config import find_policy_dir, get_audit_db_path, load_mode +from ..output import error, heading, info, panel, plain, print_mode, success, table + + +def register(subparsers): + p = subparsers.add_parser("status", help="Show current firewall status") + p.set_defaults(func=run) + + +def run(args): + lines = [] + + # Version + lines.append(f"agsec v{__version__}") + lines.append("") + + # Mode + mode = load_mode() + mode_display = mode.upper() + if mode == "halt": + mode_display += " (all actions blocked)" + elif mode == "observe": + mode_display += " (logging only, not blocking)" + elif mode == "enforce": + mode_display += " (active)" + lines.append(f" Mode: {mode_display}") + + # Installed platforms + platforms = _detect_platforms() + if platforms: + lines.append(f" Platforms: {', '.join(platforms)}") + else: + lines.append(" Platforms: none installed") + + # Policies + try: + policy_dir = find_policy_dir() + files = sorted( + glob.glob(os.path.join(policy_dir, "*.yaml")) + + glob.glob(os.path.join(policy_dir, "*.yml")) + ) + # Count statements + stmt_count = 0 + for f in files: + with open(f, "r") as fh: + import yaml + parsed = yaml.safe_load(fh.read()) + if isinstance(parsed, dict) and "statements" in parsed: + stmts = parsed.get("statements", []) + if isinstance(stmts, list): + stmt_count += len(stmts) + lines.append(f" Policies: {len(files)} files, {stmt_count} statements") + except (FileNotFoundError, Exception): + lines.append(" Policies: not configured (run `agsec init`)") + + # Audit stats + try: + audit = AuditStore(get_audit_db_path()) + stats = audit.get_execution_stats() + total = stats.get("total_executions", 0) + blocked = stats.get("blocked", 0) + if total > 0: + if mode == "observe": + lines.append(f" Audit: {total} actions logged ({blocked} would block)") + else: + lines.append(f" Audit: {total} actions logged ({blocked} blocked)") + + # Last blocked action + last_blocked = audit.get_executions(limit=100) + for ex in last_blocked: + if ex.get("policy_status") == "block": + action = ex.get("action", "unknown") + reason = ex.get("policy_reason", "") + ts = ex.get("timestamp", "") + ago = _time_ago(ts) + detail = f"{action}" + if reason: + detail += f" ({reason})" + if ago: + lines.append(f" Last block: {ago} — {detail}") + else: + lines.append(f" Last block: {detail}") + break + else: + if blocked == 0: + lines.append(" Last block: none") + else: + lines.append(" Audit: no actions logged yet") + audit.close() + except Exception: + lines.append(" Audit: database not available") + + # Output + content = "\n".join(lines) + try: + from ..output import RICH + if RICH: + panel("agsec status", content) + else: + print(content) + except Exception: + print(content) + + +def _detect_platforms(): + """Detect which platforms have agsec hooks installed.""" + platforms = [] + cwd = os.getcwd() + + checks = [ + ("claude-code", os.path.join(cwd, ".claude", "settings.json"), "agsec"), + ("codex", os.path.join(cwd, ".codex", "hooks.json"), "agsec"), + ("cursor", os.path.join(cwd, ".cursor", "hooks.json"), "agsec"), + ("windsurf", os.path.join(cwd, ".windsurf", "settings.json"), "agsec"), + ("cline", os.path.join(cwd, ".clinerules", "hooks", "agsec-check.sh"), None), + ("copilot", os.path.join(cwd, ".github", "hooks", "hooks.json"), "agsec"), + ] + + for name, path, search_str in checks: + if not os.path.isfile(path): + continue + if search_str is None: + # File existence = installed (cline) + platforms.append(name) + else: + try: + with open(path, "r") as f: + if search_str in f.read(): + platforms.append(name) + except Exception: + pass + + return platforms + + +def _time_ago(iso_timestamp): + """Convert ISO timestamp to human-readable 'X ago' string.""" + if not iso_timestamp: + return "" + try: + from datetime import datetime + ts = datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00").split("+")[0]) + now = datetime.utcnow() + diff = now - ts + seconds = int(diff.total_seconds()) + if seconds < 60: + return f"{seconds}s ago" + minutes = seconds // 60 + if minutes < 60: + return f"{minutes}m ago" + hours = minutes // 60 + if hours < 24: + return f"{hours}h ago" + days = hours // 24 + return f"{days}d ago" + except Exception: + return "" diff --git a/agsec/cli/commands/validate.py b/agsec/cli/commands/validate.py index 84af3eb..7dc85a5 100644 --- a/agsec/cli/commands/validate.py +++ b/agsec/cli/commands/validate.py @@ -7,6 +7,7 @@ from ...policy import PolicyEngine from ..config import find_policy_dir +from ..output import error, info, success def register(subparsers): @@ -20,8 +21,9 @@ def run(args): if not path: try: path = find_policy_dir() - except FileNotFoundError as e: - print(f"Error: {e}", file=sys.stderr) + except FileNotFoundError: + error("No policies directory found.") + info("Run 'agsec init' to create default policies.") sys.exit(1) engine = PolicyEngine() @@ -30,20 +32,20 @@ def run(args): results = engine.validate_directory(path) if not results: files = [f for f in os.listdir(path) if f.endswith((".yaml", ".yml"))] - print(f"All {len(files)} policy file(s) valid.") + success(f"All {len(files)} policy file(s) valid.") sys.exit(0) for filename, issues in results.items(): for issue in issues: - print(f" {filename}: {issue}", file=sys.stderr) + error(f"{filename}: {issue}") sys.exit(1) elif os.path.isfile(path): issues = engine.validate_file(path) if not issues: - print(f"Valid: {path}") + success(f"Valid: {path}") sys.exit(0) for issue in issues: - print(f" {issue}", file=sys.stderr) + error(issue) sys.exit(1) else: - print(f"Error: Path not found: {path}", file=sys.stderr) + error(f"Path not found: {path}") sys.exit(1) diff --git a/agsec/cli/main.py b/agsec/cli/main.py index 8b14936..5ad02c7 100644 --- a/agsec/cli/main.py +++ b/agsec/cli/main.py @@ -7,13 +7,16 @@ def main(): + from .. import __version__ + parser = argparse.ArgumentParser( prog="agsec", description="AI Agent Action Firewall — enforce policies on agent actions", ) + parser.add_argument("--version", action="version", version=f"agsec {__version__}") subparsers = parser.add_subparsers(dest="command") - from .commands import analyze, audit, check, init, install, mode, policy, validate + from .commands import analyze, audit, check, init, install, mode, policy, status, validate init.register(subparsers) check.register(subparsers) @@ -23,6 +26,7 @@ def main(): policy.register(subparsers) audit.register(subparsers) analyze.register(subparsers) + status.register(subparsers) mode.register_observe(subparsers) mode.register_enforce(subparsers) mode.register_halt(subparsers) diff --git a/agsec/cli/output.py b/agsec/cli/output.py new file mode 100644 index 0000000..1859bda --- /dev/null +++ b/agsec/cli/output.py @@ -0,0 +1,182 @@ +"""agsec CLI output utilities — Rich if available, plain text fallback.""" + +from __future__ import annotations + +import sys +from typing import Any, List, Optional + +try: + from rich.console import Console + from rich.panel import Panel + from rich.table import Table as RichTable + from rich.text import Text + + RICH = True + console = Console(stderr=False) + err_console = Console(stderr=True) +except ImportError: + RICH = False + console = None + err_console = None + + +# --------------------------------------------------------------------------- +# Core message functions +# --------------------------------------------------------------------------- + + +def success(msg: str) -> None: + if RICH: + console.print(f"[green]{msg}[/green]") + else: + print(f"[+] {msg}") + + +def error(msg: str) -> None: + if RICH: + err_console.print(f"[bold red]{msg}[/bold red]") + else: + print(f"[X] {msg}", file=sys.stderr) + + +def warn(msg: str) -> None: + if RICH: + console.print(f"[yellow]{msg}[/yellow]") + else: + print(f"[!] {msg}") + + +def info(msg: str) -> None: + if RICH: + console.print(f"[dim]{msg}[/dim]") + else: + print(f" {msg}") + + +def plain(msg: str) -> None: + if RICH: + console.print(msg) + else: + print(msg) + + +# --------------------------------------------------------------------------- +# Status-colored output +# --------------------------------------------------------------------------- + + +def status_allow(msg: str) -> None: + if RICH: + console.print(f" [green]+[/green] {msg}") + else: + print(f" [+] {msg}") + + +def status_block(msg: str) -> None: + if RICH: + console.print(f" [red]X[/red] {msg}") + else: + print(f" [X] {msg}") + + +def status_review(msg: str) -> None: + if RICH: + console.print(f" [yellow]?[/yellow] {msg}") + else: + print(f" [?] {msg}") + + +# --------------------------------------------------------------------------- +# Headings and structure +# --------------------------------------------------------------------------- + + +def heading(msg: str) -> None: + if RICH: + console.print(f"\n[bold]{msg}[/bold]") + else: + print(f"\n{msg}") + print("=" * len(msg)) + + +def subheading(msg: str) -> None: + if RICH: + console.print(f"[bold]{msg}[/bold]") + else: + print(msg) + + +def panel(title: str, content: str) -> None: + if RICH: + console.print(Panel(content, title=title, border_style="blue")) + else: + print(f"--- {title} ---") + print(content) + print() + + +# --------------------------------------------------------------------------- +# Tables +# --------------------------------------------------------------------------- + + +def table(headers: List[str], rows: List[List[str]], title: Optional[str] = None) -> None: + if RICH: + t = RichTable(title=title, show_header=True, header_style="bold") + for h in headers: + t.add_column(h) + for row in rows: + t.add_row(*[str(c) for c in row]) + console.print(t) + else: + if title: + print(f"\n{title}") + # Simple column formatting + if not rows: + return + widths = [len(h) for h in headers] + for row in rows: + for i, cell in enumerate(row): + if i < len(widths): + widths[i] = max(widths[i], len(str(cell))) + fmt = " ".join(f"{{:<{w}}}" for w in widths) + print(fmt.format(*headers)) + print(fmt.format(*["-" * w for w in widths])) + for row in rows: + padded = [str(row[i]) if i < len(row) else "" for i in range(len(headers))] + print(fmt.format(*padded)) + + +# --------------------------------------------------------------------------- +# Mode display +# --------------------------------------------------------------------------- + + +def mode_label(mode: str) -> str: + if not RICH: + return mode.upper() + colors = {"observe": "yellow", "enforce": "green", "halt": "red"} + color = colors.get(mode, "white") + return f"[{color}]{mode.upper()}[/{color}]" + + +def print_mode(mode: str) -> None: + if RICH: + icons = {"observe": "[yellow]![/yellow]", "enforce": "[green]+[/green]", "halt": "[red]X[/red]"} + icon = icons.get(mode, " ") + console.print(f" {icon} Mode: {mode_label(mode)}") + else: + print(f" Mode: {mode.upper()}") + + +# --------------------------------------------------------------------------- +# Severity colors (for analyze) +# --------------------------------------------------------------------------- + + +def severity_label(severity: str) -> str: + if not RICH: + return severity.upper() + colors = {"critical": "bold red", "high": "red", "medium": "yellow", "low": "blue"} + color = colors.get(severity, "white") + return f"[{color}]{severity.upper()}[/{color}]" diff --git a/agsec/integrations/anthropic.py b/agsec/integrations/anthropic.py index 75e761e..09e4b59 100644 --- a/agsec/integrations/anthropic.py +++ b/agsec/integrations/anthropic.py @@ -14,6 +14,7 @@ from __future__ import annotations +import json from typing import Any, Optional from ..types import PolicyStatus @@ -27,6 +28,70 @@ from ._base import build_engine, check_tool +class _GuardedAnthropicStream: + """Wraps Anthropic stream, checks tool calls after completion. + + Yields all events transparently. After iteration completes, + check ._agsec_blocked for any tool calls that violate policy. + """ + + def __init__(self, stream, engine): + self._stream = stream + self._engine = engine + self._tool_blocks = {} + self._current_index = -1 + self._agsec_blocked = [] + + def __iter__(self): + for event in self._stream: + self._accumulate(event) + yield event + self._check_policy() + + def __enter__(self): + if hasattr(self._stream, "__enter__"): + self._stream.__enter__() + return self + + def __exit__(self, *args): + if hasattr(self._stream, "__exit__"): + return self._stream.__exit__(*args) + + def _accumulate(self, event): + event_type = getattr(event, "type", "") + if event_type == "content_block_start": + cb = getattr(event, "content_block", None) + if cb and getattr(cb, "type", "") == "tool_use": + self._current_index = getattr(event, "index", -1) + self._tool_blocks[self._current_index] = { + "id": cb.id, + "name": cb.name, + "input_json": "", + } + elif event_type == "content_block_delta": + delta = getattr(event, "delta", None) + if delta and getattr(delta, "type", "") == "input_json_delta": + idx = getattr(event, "index", self._current_index) + if idx in self._tool_blocks: + self._tool_blocks[idx]["input_json"] += delta.partial_json + + def _check_policy(self): + for idx, tb in self._tool_blocks.items(): + try: + input_params = json.loads(tb["input_json"]) if tb["input_json"] else {} + except (json.JSONDecodeError, TypeError): + input_params = {} + status, reason, metadata = check_tool(self._engine, tb["name"], input_params) + if status != PolicyStatus.ALLOW: + self._agsec_blocked.append({ + "tool_use_id": tb["id"], + "name": tb["name"], + "status": status.value, + "reason": reason, + "sid": metadata.get("sid", ""), + }) + + def protect( client: Any, *rules: ToolRule, @@ -55,10 +120,8 @@ def protect( def wrapped_create(*args, **kwargs): if kwargs.get("stream", False): - raise NotImplementedError( - "agsec does not yet support streaming responses. " - "Use stream=False or call the original client directly." - ) + stream = original_create(*args, **kwargs) + return _GuardedAnthropicStream(stream, engine) response = original_create(*args, **kwargs) diff --git a/agsec/integrations/openai.py b/agsec/integrations/openai.py index fe4a202..5e84e9e 100644 --- a/agsec/integrations/openai.py +++ b/agsec/integrations/openai.py @@ -31,6 +31,72 @@ from ._base import build_engine, check_tool +class _GuardedStream: + """Wraps OpenAI Stream, checks tool calls after completion. + + Yields all chunks transparently. After iteration completes, + check ._agsec_blocked for any tool calls that violate policy. + """ + + def __init__(self, stream, engine): + self._stream = stream + self._engine = engine + self._accumulated = {} + self._agsec_blocked = [] + + def __iter__(self): + for chunk in self._stream: + self._accumulate(chunk) + yield chunk + self._check_policy() + + def __enter__(self): + if hasattr(self._stream, "__enter__"): + self._stream.__enter__() + return self + + def __exit__(self, *args): + if hasattr(self._stream, "__exit__"): + return self._stream.__exit__(*args) + + @property + def response(self): + return getattr(self._stream, "response", None) + + def _accumulate(self, chunk): + if not hasattr(chunk, "choices") or not chunk.choices: + return + delta = chunk.choices[0].delta if hasattr(chunk.choices[0], "delta") else None + if not delta: + return + tool_calls = getattr(delta, "tool_calls", None) + if not tool_calls: + return + for tc in tool_calls: + idx = tc.index + if idx not in self._accumulated: + self._accumulated[idx] = {"id": "", "name": "", "arguments": ""} + if tc.id: + self._accumulated[idx]["id"] = tc.id + if tc.function: + if tc.function.name: + self._accumulated[idx]["name"] = tc.function.name + if tc.function.arguments: + self._accumulated[idx]["arguments"] += tc.function.arguments + + def _check_policy(self): + for idx, tc in self._accumulated.items(): + status, reason, metadata = check_tool(self._engine, tc["name"], tc["arguments"]) + if status != PolicyStatus.ALLOW: + self._agsec_blocked.append({ + "tool_call_id": tc["id"], + "name": tc["name"], + "status": status.value, + "reason": reason, + "sid": metadata.get("sid", ""), + }) + + def protect( client: Any, *rules: ToolRule, @@ -62,10 +128,8 @@ def protect( def wrapped_create(*args, **kwargs): if kwargs.get("stream", False): - raise NotImplementedError( - "agsec does not yet support streaming responses. " - "Use stream=False or call the original client directly." - ) + stream = original_create(*args, **kwargs) + return _GuardedStream(stream, engine) response = original_create(*args, **kwargs) diff --git a/agsec/policy/conditions.py b/agsec/policy/conditions.py index 3a7e849..2ad46a9 100644 --- a/agsec/policy/conditions.py +++ b/agsec/policy/conditions.py @@ -52,7 +52,7 @@ def _evaluate_condition(value: Any, condition: Any) -> bool: return str(value).endswith(str(expected)) if op == "regex": pattern = str(expected) - if len(pattern) > 1000: + if len(pattern) > 500: return False # Reject overly long patterns try: return bool(re.search(pattern, str(value))) diff --git a/agsec/policy/normalizers.py b/agsec/policy/normalizers.py new file mode 100644 index 0000000..c0ac83c --- /dev/null +++ b/agsec/policy/normalizers.py @@ -0,0 +1,58 @@ +"""Input normalization to prevent encoding bypasses. + +Decodes common encodings (base64, unicode escapes, hex escapes) +before policy evaluation, so regex patterns catch obfuscated inputs. +""" + +from __future__ import annotations + +import base64 +import binascii +import re + + +def normalize_value(value): + """Normalize a string value by decoding common encodings. + + Returns original value if not a string or decoding fails. + Non-destructive: only decodes if the result is valid printable text. + """ + if not isinstance(value, str): + return value + + normalized = value + + # Try unicode escapes (\u0052 -> R) + if "\\u" in normalized: + try: + decoded = normalized.encode("utf-8").decode("unicode_escape") + if decoded != normalized and decoded.isprintable(): + normalized = decoded + except (UnicodeDecodeError, ValueError): + pass + + # Try hex escapes (\x72 -> r) + if "\\x" in normalized: + try: + decoded = normalized.encode("utf-8").decode("unicode_escape") + if decoded != normalized and decoded.isprintable(): + normalized = decoded + except (UnicodeDecodeError, ValueError): + pass + + # Try base64 — only if the entire string looks like pure base64 + # (no spaces, only base64 chars, reasonable length) + stripped = normalized.strip() + if ( + len(stripped) >= 4 + and len(stripped) <= 1000 + and re.match(r"^[A-Za-z0-9+/=]+$", stripped) + ): + try: + decoded = base64.b64decode(stripped, validate=True).decode("utf-8") + if decoded.isprintable() and len(decoded) >= 2: + normalized = decoded + except (ValueError, UnicodeDecodeError, binascii.Error): + pass + + return normalized diff --git a/agsec/policy/resolvers.py b/agsec/policy/resolvers.py index e847f1b..b2aeada 100644 --- a/agsec/policy/resolvers.py +++ b/agsec/policy/resolvers.py @@ -2,6 +2,8 @@ from typing import Any, Dict, List, Optional +from .normalizers import normalize_value + def _deep_get(obj: Any, keys: List[str]) -> Any: """Walk a nested dict by key path. Returns None if any segment is missing.""" @@ -14,18 +16,22 @@ def _deep_get(obj: Any, keys: List[str]) -> Any: def _resolve_value(key: str, params: Dict[str, Any], context: Optional[Dict[str, Any]]) -> Any: - """Resolve a dotted key like 'params.user.address.country' from params or context.""" + """Resolve a dotted key like 'params.user.address.country' from params or context. + + String values are normalized to decode common encodings (base64, unicode, hex) + before being returned, preventing policy bypass via obfuscation. + """ parts = key.split(".") if parts[0] == "params": - return _deep_get(params, parts[1:]) + return normalize_value(_deep_get(params, parts[1:])) if parts[0] == "context": - return _deep_get(context or {}, parts[1:]) + return normalize_value(_deep_get(context or {}, parts[1:])) # Legacy: bare key — try params first, then context result = _deep_get(params, parts) if result is not None: - return result + return normalize_value(result) if context: - return _deep_get(context, parts) + return normalize_value(_deep_get(context, parts)) return None diff --git a/agsec/templates/policies/01_base.yaml b/agsec/templates/policies/01_base.yaml index 3d79b7c..2916e8e 100644 --- a/agsec/templates/policies/01_base.yaml +++ b/agsec/templates/policies/01_base.yaml @@ -8,7 +8,17 @@ statements: actions: - "internal.*" - # Allow all read/search operations — low risk + # Block reads of sensitive files (secrets, credentials, keys) + - sid: "BlockReadSecrets" + effect: deny + actions: ["file.read"] + conditions: + params.file_path: + op: "regex" + value: "(\\.env$|\\.env\\..+|credentials\\.json|secrets\\.ya?ml|\\.ssh/|id_rsa|\\.aws/credentials|\\.gcloud/|service[_-]account.*\\.json)" + reason: "Reading secret files is blocked" + + # Allow all other read/search operations — low risk - sid: "AllowReadOps" effect: allow actions: diff --git a/agsec/templates/policies/02_bash.yaml b/agsec/templates/policies/02_bash.yaml index 5917419..3a99e64 100644 --- a/agsec/templates/policies/02_bash.yaml +++ b/agsec/templates/policies/02_bash.yaml @@ -21,15 +21,35 @@ statements: value: "(chmod\\s+777|mkfs\\.|dd\\s+if=|shred\\s|wipe\\s)" reason: "Destructive filesystem command blocked" - # Block destructive database commands - - sid: "BlockDestructiveSQL" + # Block DDL commands (DROP, TRUNCATE, ALTER DROP) + - sid: "BlockDDL" effect: deny actions: ["bash.execute"] conditions: params.command: op: "regex" - value: "(?i)(DROP\\s+(TABLE|DATABASE)|TRUNCATE\\s+TABLE|DELETE\\s+FROM\\s+\\S+\\s*$)" - reason: "Destructive database command blocked" + value: "(?i)(DROP\\s+(TABLE|DATABASE)|TRUNCATE\\s+TABLE|ALTER\\s+TABLE\\s+\\S+\\s+DROP)" + reason: "Destructive DDL command blocked — agents should not modify database schema" + + # Block DML commands (DELETE, UPDATE, INSERT, MERGE) + - sid: "BlockDML" + effect: deny + actions: ["bash.execute"] + conditions: + params.command: + op: "regex" + value: "(?i)(DELETE\\s+FROM|UPDATE\\s+\\S+\\s+SET|INSERT\\s+INTO|MERGE\\s+INTO)" + reason: "Direct SQL data modification blocked — use application layer for data changes" + + # Block direct access to agsec audit database + - sid: "BlockAuditDBAccess" + effect: deny + actions: ["bash.execute"] + conditions: + params.command: + op: "regex" + value: "(sqlite3|psql|mysql|mongosh?)\\s+.*(\\.agsec/|audit\\.db)" + reason: "Direct access to audit database is blocked — protects audit trail integrity" # Block direct access to secrets via bash - sid: "BlockSecretAccess" @@ -51,6 +71,16 @@ statements: value: "(curl|wget|nc|ncat).*(-d|--data|--upload|-T).*(\\.env|credentials|secret|password|token)" reason: "Potential data exfiltration blocked" + # Block encoded command execution (base64/hex decode piped to shell) + - sid: "BlockEncodedExecution" + effect: deny + actions: ["bash.execute"] + conditions: + params.command: + op: "regex" + value: "(base64\\s+-d|xxd\\s+-r|python[23]?\\s+-c\\s+['\"]*(exec|eval|import)|perl\\s+-e|ruby\\s+-e).*(\\||;)\\s*(sh|bash|zsh|exec)" + reason: "Encoded command execution blocked — potential policy bypass" + # Allow other bash commands - sid: "AllowBash" effect: allow diff --git a/agsec/templates/policies/03_files.yaml b/agsec/templates/policies/03_files.yaml index 10fe233..162a89a 100644 --- a/agsec/templates/policies/03_files.yaml +++ b/agsec/templates/policies/03_files.yaml @@ -18,7 +18,7 @@ statements: conditions: params.file_path: op: "regex" - value: "(\\.agsec\\.yaml|policies/.*\\.ya?ml|\\.claude/settings.*\\.json|\\.codex/hooks\\.json)" + value: "(\\.agsec\\.yaml|policies/.*\\.ya?ml|\\.claude/settings.*\\.json|\\.codex/hooks\\.json|\\.cursor/hooks\\.json|\\.windsurf/settings.*\\.json|\\.clinerules/hooks/|\\.github/hooks/hooks\\.json|\\.copilot/hooks/hooks\\.json)" reason: "Writing to agsec policies or hook config is blocked" # Block writes to system directories diff --git a/agsec/threat.py b/agsec/threat.py index b25c899..7d7378e 100644 --- a/agsec/threat.py +++ b/agsec/threat.py @@ -53,6 +53,19 @@ class ThreatReport: THREAT_PATTERNS: List[ThreatPattern] = [ # CRITICAL + ThreatPattern( + id="read_secrets", + name="Reading secret files", + severity=Severity.CRITICAL, + action_types=["file.read"], + param_field="file_path", + regex=r"(\.env$|\.env\..+|credentials\.json|secrets\.ya?ml|\.ssh/|id_rsa|\.aws/credentials|\.gcloud/|service[_-]account.*\.json)", + consequence=( + "Secrets exposed to agent context \u2014 API keys, credentials, or private keys " + "readable by the agent and potentially exfiltrated in subsequent actions" + ), + recommendation="Add a deny rule for file.read on sensitive file patterns", + ), ThreatPattern( id="secret_access", name="Secret access via bash", @@ -79,19 +92,58 @@ class ThreatReport: ), recommendation="Ensure BlockDataExfiltration policy (02_bash.yaml) is enforced", ), + ThreatPattern( + id="encoded_execution", + name="Encoded command execution", + severity=Severity.CRITICAL, + action_types=["bash.execute"], + param_field="command", + regex=r"(base64\s+-d|xxd\s+-r|python[23]?\s+-c\s+['\"]*(exec|eval|import)|perl\s+-e|ruby\s+-e).*(\|)\s*(sh|bash|zsh|exec)", + consequence=( + "Agent attempting to decode and execute hidden commands \u2014 " + "likely policy bypass attempt" + ), + recommendation="Investigate: agent used encoding to hide its true intent", + ), ThreatPattern( id="destructive_sql", - name="Destructive SQL", + name="Destructive SQL (DDL)", severity=Severity.CRITICAL, action_types=["bash.execute"], param_field="command", - regex=r"(?i)(DROP\s+(TABLE|DATABASE)|TRUNCATE\s+TABLE|DELETE\s+FROM\s+\S+\s*$)", + regex=r"(?i)(DROP\s+(TABLE|DATABASE)|TRUNCATE\s+TABLE|ALTER\s+TABLE\s+\S+\s+DROP)", consequence=( "Irreversible database destruction \u2014 tables or entire databases " "dropped with no recovery without backups" ), recommendation="Ensure BlockDestructiveSQL policy (02_bash.yaml) is enforced", ), + ThreatPattern( + id="dml_sql", + name="Data modification SQL (DML)", + severity=Severity.HIGH, + action_types=["bash.execute"], + param_field="command", + regex=r"(?i)(DELETE\s+FROM|UPDATE\s+\S+\s+SET|INSERT\s+INTO|MERGE\s+INTO)", + consequence=( + "Direct database modification via SQL \u2014 data corruption, " + "unauthorized changes, or audit trail manipulation" + ), + recommendation="Review raw SQL commands — use application layer for data changes", + ), + ThreatPattern( + id="audit_tampering", + name="Audit database access", + severity=Severity.CRITICAL, + action_types=["bash.execute"], + param_field="command", + regex=r"(sqlite3|psql|mysql|mongosh?)\s+.*(\\.agsec/|audit\.db)", + consequence=( + "Direct access to audit database \u2014 attacker could read, modify, " + "or wipe the audit trail to cover tracks" + ), + recommendation="URGENT: Agent accessed audit database directly \u2014 investigate immediately", + ), # HIGH ThreatPattern( id="file_deletion", @@ -119,6 +171,45 @@ class ThreatReport: ), recommendation="Ensure BlockDestructiveFS policy (02_bash.yaml) is enforced", ), + ThreatPattern( + id="read_system", + name="Reading system files", + severity=Severity.HIGH, + action_types=["file.read"], + param_field="file_path", + regex=r"^/(etc|proc|sys|boot|root)/", + consequence=( + "System file contents exposed \u2014 /etc/passwd, /etc/shadow, or process info " + "leaking to agent context" + ), + recommendation="Add a deny rule for file.read on system directories", + ), + ThreatPattern( + id="read_policy", + name="Reading policy/guardrail config", + severity=Severity.HIGH, + action_types=["file.read"], + param_field="file_path", + regex=r"(\.agsec\.yaml|policies/.*\.ya?ml|\.claude/settings.*\.json|\.codex/hooks\.json|\.cursor/hooks\.json|\.windsurf/settings.*\.json|\.clinerules/hooks/|\.github/hooks/hooks\.json|\.copilot/hooks/hooks\.json)", + consequence=( + "Agent reading its own guardrails \u2014 could learn what's blocked " + "and craft bypass strategies" + ), + recommendation="Add a deny rule for file.read on policy and hook config files", + ), + ThreatPattern( + id="notebook_sensitive", + name="Notebook edit of sensitive file", + severity=Severity.HIGH, + action_types=["notebook.edit"], + param_field="file_path", + regex=r"(\.env$|\.env\..+|credentials\.json|secrets\.ya?ml|\.ssh/|id_rsa|\.aws/credentials)", + consequence=( + "Credential tampering via notebook \u2014 same risk as direct file.write " + "to sensitive files" + ), + recommendation="Add notebook.edit to the BlockWriteSensitive policy in 03_files.yaml", + ), ThreatPattern( id="write_sensitive", name="Write to sensitive files", @@ -151,7 +242,7 @@ class ThreatReport: severity=Severity.HIGH, action_types=["file.write", "file.edit"], param_field="file_path", - regex=r"(\.agsec\.yaml|policies/.*\.ya?ml|\.claude/settings.*\.json|\.codex/hooks\.json)", + regex=r"(\.agsec\.yaml|policies/.*\.ya?ml|\.claude/settings.*\.json|\.codex/hooks\.json|\.cursor/hooks\.json|\.windsurf/settings.*\.json|\.clinerules/hooks/|\.github/hooks/hooks\.json|\.copilot/hooks/hooks\.json)", consequence=( "Policy escape \u2014 agent modifying its own guardrails " "could disable all protections" @@ -198,6 +289,19 @@ class ThreatReport: ), recommendation="Ensure BlockPushProtected policy (05_git.yaml) is enforced", ), + ThreatPattern( + id="scan_secrets", + name="Scanning for secrets", + severity=Severity.MEDIUM, + action_types=["file.glob", "file.grep"], + param_field="pattern", + regex=r"(\.env|credentials|secret|password|private_key|id_rsa|\.aws|\.ssh|token)", + consequence=( + "Agent actively searching for secret files \u2014 " + "reconnaissance step before exfiltration" + ), + recommendation="Monitor file.glob and file.grep patterns for sensitive keywords", + ), ThreatPattern( id="external_fetch", name="External web fetch", @@ -246,6 +350,7 @@ def classify(self, executions: List[Dict[str, Any]]) -> ThreatReport: for row in executions: action = row.get("action", "") policy_status = row.get("policy_status", "allow") + outcome = row.get("outcome") # "allowed", "blocked", "review", or None (legacy) # Parse params raw_params = row.get("params", "{}") @@ -273,10 +378,17 @@ def classify(self, executions: List[Dict[str, Any]]) -> ThreatReport: policy_status=policy_status, ) - if policy_status == "block": + # Use outcome if available (new logs), fall back to policy_status (legacy) + if outcome: + actually_blocked = outcome == "blocked" + else: + # Legacy rows: no outcome column, use policy_status + actually_blocked = policy_status == "block" + + if actually_blocked: blocked.append(finding) else: - # "allow" and "review" are real threats + # "allowed" and "review" = action got through = threat threats.append(finding) # Calculate blast radius from threats only diff --git a/docs/cli.md b/docs/cli.md index 5755153..7ad6c3f 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -12,20 +12,38 @@ agsec init --observe # observe mode (audit only, no blocking) agsec init --dir .agsec/policies # custom directory ``` -### `agsec install claude-code|codex` +### `agsec install ` Activate the firewall for an agent platform. ```bash -agsec install claude-code -agsec install codex +agsec install claude-code # Claude Code + Claude Cowork +agsec install codex # OpenAI Codex +agsec install cursor # Cursor +agsec install windsurf # Windsurf (Codeium) +agsec install cline # Cline +agsec install copilot # GitHub Copilot (project + user level) ``` +### `agsec uninstall ` + +Remove the agsec hook from a platform. + +## Status + +### `agsec status` + +Show firewall status at a glance — mode, installed platforms, policy count, audit stats, last blocked action. + +### `agsec --version` + +Show the installed agsec version. + ## Policy Management ### `agsec policy list` -Show all active policy statements across all files. +Show all active policy statements across all files, with colored allow/deny/review markers. ### `agsec policy add` @@ -58,12 +76,20 @@ agsec validate policy.yaml # single file ### `agsec observe` -Switch to observe mode. All actions are allowed but every policy decision is logged. Use `agsec audit --stats` to see what would be blocked. +Switch to observe mode. All actions are allowed but every policy decision is logged with its actual outcome. Use `agsec audit --stats` to see what would be blocked. ### `agsec enforce` Switch to enforce mode. Policies are enforced — blocked actions are denied. +### `agsec halt` + +Kill switch — immediately block ALL agent actions. Saves previous mode for resume. + +### `agsec resume` + +Restore from halt — returns to the mode that was active before halt. + ## Audit ### `agsec audit [--stats] [--action NAME] [--limit N] [--json]` @@ -78,3 +104,52 @@ agsec audit --json # machine-readable output ``` In observe mode, stats show "would block" instead of "blocked". + +### `agsec audit --prune [DAYS]` + +Delete audit records older than N days. Default: 7 days. + +```bash +agsec audit --prune # delete records older than 7 days +agsec audit --prune 30 # delete records older than 30 days +``` + +### `agsec audit --clear` + +Delete ALL audit records. + +You can also auto-prune by setting the environment variable: +```bash +export AGSEC_AUDIT_RETENTION_DAYS=7 +``` + +## Threat Analysis + +### `agsec analyze [--hours N] [--days N] [--json]` + +Threat analysis with blast radius scoring. Classifies audit logs against 22 built-in threat patterns. Shows severity (CRITICAL/HIGH/MEDIUM/LOW), consequences, and recommendations. + +```bash +agsec analyze # all time +agsec analyze --hours 24 # last 24 hours +agsec analyze --days 7 # last 7 days +agsec analyze --json # machine-readable output +``` + +### `agsec analyze --all` + +Full activity report — every action grouped by human-readable type (Shell Commands, File Reads, Web Requests, etc.) with actual values. Blocked/review items sorted to top. + +```bash +agsec analyze --all # all activity +agsec analyze --all --hours 1 # last hour +agsec analyze --all --json # JSON output +``` + +## Environment Variables + +| Variable | Description | Default | +|----------|-------------|---------| +| `AGSEC_MODE` | Override mode (observe/enforce) | from `.agsec.yaml` | +| `AGSEC_AUDIT_DB` | Override audit database path | `~/.agsec/audit.db` | +| `AGSEC_AUDIT_RETENTION_DAYS` | Auto-prune records older than N days | disabled | diff --git a/docs/integrations.md b/docs/integrations.md index fc56635..f3ca749 100644 --- a/docs/integrations.md +++ b/docs/integrations.md @@ -2,7 +2,9 @@ agsec works with any agent platform. One policy engine, same audit trail everywhere. -## Claude Code +## System Agents (Hook-Based) + +### Claude Code ```bash pip install agsec @@ -10,17 +12,47 @@ agsec init agsec install claude-code ``` -Enforcement is at the runtime level. The agent cannot bypass it. +Enforcement is at the runtime level. The agent cannot bypass it. Works with Claude Code and Claude Cowork. -## OpenAI Codex +### OpenAI Codex ```bash -pip install agsec -agsec init agsec install codex ``` -## LangChain +### Cursor + +```bash +agsec install cursor +``` + +### Windsurf + +```bash +agsec install windsurf +``` + +### Cline + +```bash +agsec install cline +``` + +### GitHub Copilot + +```bash +agsec install copilot +``` + +Installs to both project-level (`.github/hooks/`) and user-level (`~/.copilot/hooks/`). The project-level hook must be committed for the cloud coding agent. + +All platforms embed `--agent ` in the hook command, enabling agent-specific policy layers at `~/.agsec/agents/{agent-name}/`. + +--- + +## Python SDK Integrations + +### LangChain ```bash pip install agsec[langchain] @@ -54,7 +86,7 @@ agent = create_react_agent(llm, guard( )) ``` -## OpenAI SDK +### OpenAI SDK Works with OpenAI, OpenRouter, Groq, Together, Fireworks — anything OpenAI-compatible. @@ -82,11 +114,25 @@ client = protect(OpenAI(base_url="https://openrouter.ai/api/v1", api_key="..."), ) ``` -Blocked tool calls are filtered from the response. Check `response._agsec_blocked` for details. +**Non-streaming:** Blocked tool calls are filtered from the response. Check `response._agsec_blocked` for details. -**Note:** Streaming (`stream=True`) is not yet supported. Use `stream=False`. +**Streaming:** Chunks flow through unchanged (zero latency impact). Tool calls are accumulated and checked after the stream completes. Check `stream._agsec_blocked` after consuming the stream. -## Anthropic SDK +**Important:** Policy checks run only after the full stream is consumed. If you break out of the iteration early (e.g., `for chunk in stream: break`), policy enforcement will not run. Always consume the full stream for enforcement to apply. + +```python +stream = client.chat.completions.create(model="gpt-4", messages=[...], tools=[...], stream=True) + +for chunk in stream: + # Chunks arrive in real-time + ... + +# After stream completes, check blocked tool calls +if stream._agsec_blocked: + print(f"Blocked: {stream._agsec_blocked}") +``` + +### Anthropic SDK ```bash pip install agsec[anthropic] @@ -104,9 +150,9 @@ client = protect(Anthropic(), response = client.messages.create(model="claude-sonnet-4-20250514", messages=[...], tools=[...]) ``` -Same behavior as OpenAI — blocked `tool_use` blocks are filtered from the response. +Same behavior as OpenAI — blocked `tool_use` blocks are filtered from the response. Text blocks are always preserved. Streaming supported with the same `stream._agsec_blocked` pattern. Same limitation applies: always consume the full stream for enforcement to run. -## Generic Python (any framework) +### Generic Python (any framework) ```python from agsec import guard @@ -122,8 +168,11 @@ async def charge(amount, recipient): Works with sync and async functions. Raises `PolicyViolationError` on block. +--- + ## Install All ```bash -pip install agsec[all] # openai + anthropic + langchain +pip install agsec[all] # openai + anthropic + langchain + rich +pip install agsec[rich] # colored CLI output (optional) ``` diff --git a/docs/observe-mode.md b/docs/observe-mode.md index 21fd3c0..32f92ed 100644 --- a/docs/observe-mode.md +++ b/docs/observe-mode.md @@ -1,6 +1,6 @@ # Observe Mode -Observe mode lets you deploy agsec without blocking anything. Every action is allowed, but every policy decision is logged — including what would have been blocked. +Observe mode lets you deploy agsec without blocking anything. Every action is allowed, but every policy decision is logged with its actual outcome — including what would have been blocked. ## Why @@ -32,8 +32,10 @@ agsec observe 1. agsec init --observe # deploy with logging only 2. Agent runs normally # everything is allowed 3. agsec audit --stats # see what would be blocked -4. Edit policies as needed # tune based on real data -5. agsec enforce # start blocking +4. agsec analyze # threat analysis with blast radius +5. agsec analyze --all # full activity report +6. Edit policies as needed # tune based on real data +7. agsec enforce # start blocking ``` ## Commands @@ -41,6 +43,9 @@ agsec observe ```bash agsec observe # switch to observe mode agsec enforce # switch to enforce mode +agsec halt # kill switch — block everything immediately +agsec resume # restore previous mode after halt +agsec status # see current mode at a glance ``` ## Audit in Observe Mode @@ -57,12 +62,34 @@ Audit Statistics (OBSERVE mode) The "would block" and "would review" counts show actions that policies evaluated as BLOCK/REVIEW but were allowed through because of observe mode. +## Threat Analysis + +```bash +$ agsec analyze --hours 24 +``` + +Shows blast radius, severity breakdown (CRITICAL/HIGH/MEDIUM/LOW), consequences, and recommendations. In observe mode, all findings are treated as threats (since nothing was actually blocked). + +```bash +$ agsec analyze --all +``` + +Full activity report — every action grouped by type (Shell Commands, File Reads, etc.) with actual values. Blocked/review items sorted to top. + +## Accurate Outcome Logging + +Every audit record stores both the **policy decision** (what the engine said) and the **actual outcome** (what really happened). This means: + +- Switching from observe to enforce doesn't corrupt your analysis +- `agsec analyze` accurately shows what got through vs what was blocked, regardless of mode changes + ## How It Works - A `.agsec.yaml` config file stores the current mode - `agsec check` reads this config before evaluating -- In observe mode: policies are evaluated, decisions are logged, but exit code is always 0 (allow) -- In enforce mode: policies are evaluated, decisions are logged, and blocked actions exit with non-zero +- In observe mode: policies are evaluated, decisions are logged with `outcome=allowed`, exit code is always 0 +- In enforce mode: policies are evaluated, decisions are logged with actual outcome, blocked actions exit non-zero +- In halt mode: everything is blocked immediately ## Environment Variable Override diff --git a/docs/owasp-mapping.md b/docs/owasp-mapping.md index fe24e7e..6ffc379 100644 --- a/docs/owasp-mapping.md +++ b/docs/owasp-mapping.md @@ -12,12 +12,12 @@ agsec addresses **7 of 10** OWASP Agentic risks. We are transparent about the 3 | ASI02 | Tool Misuse & Exploitation | Core feature | HIGH | | ASI03 | Identity & Privilege Abuse | Core feature | HIGH | | ASI04 | Supply Chain Vulnerabilities | Not addressed | NONE | -| ASI05 | Unexpected Code Execution | Policy gating | MEDIUM | +| ASI05 | Unexpected Code Execution | Policy gating + normalization | HIGH | | ASI06 | Memory & Context Poisoning | Not addressed | NONE | | ASI07 | Insecure Inter-Agent Comms | Not addressed | NONE | | ASI08 | Cascading Failures | Circuit breaking | HIGH | | ASI09 | Human-Agent Trust | Review + audit | MEDIUM | -| ASI10 | Rogue Agents | Kill switch + audit | HIGH | +| ASI10 | Rogue Agents | Kill switch + self-protection | HIGH | ## Detailed Mapping @@ -30,8 +30,9 @@ agsec addresses **7 of 10** OWASP Agentic risks. We are transparent about the 3 - Limits what a hijacked agent can do through default-deny policies - Every action is checked against policies regardless of the agent's intent - If an agent is tricked into running `rm -rf /`, the policy blocks it +- 22 built-in threat patterns detect dangerous actions even in observe mode -**agsec features:** Default-deny posture, per-action policy enforcement, condition-based blocking (regex, contains, etc.) +**agsec features:** Default-deny posture, per-action policy enforcement, threat analysis with blast radius scoring ### ASI02 - Tool Misuse & Exploitation (HIGH) @@ -42,9 +43,10 @@ agsec addresses **7 of 10** OWASP Agentic risks. We are transparent about the 3 - Glob pattern matching for action groups (`payment.*`, `*.delete`) - Parameter-level conditions (`deny payment when amount > 10000`) - Deep nested field access (`params.recipient.country`) +- Input normalization prevents encoding bypasses (base64, unicode, hex) - Observe mode to audit tool usage before enforcing -**agsec features:** PolicyEngine, YAML statements, condition operators, observe mode, audit trail +**agsec features:** PolicyEngine, YAML statements, condition operators, input normalization, observe mode, audit trail ### ASI03 - Identity & Privilege Abuse (HIGH) @@ -52,11 +54,12 @@ agsec addresses **7 of 10** OWASP Agentic risks. We are transparent about the 3 **What agsec does:** - Agent-scoped policies (different agents get different permissions) +- Layered policy evaluation (project + agent layers, like AWS IAM permission boundaries) - Default-deny with explicit allow (IAM semantics) - Per-action authorization, not blanket access -- Full audit trail of every action + policy decision +- Full audit trail of every action + policy decision + actual outcome -**agsec features:** Agent overlay policies (`~/.agsec/agents/{name}/`), default-deny, audit store +**agsec features:** LayeredPolicyEngine, agent overlay policies (`~/.agsec/agents/{name}/`), default-deny, outcome logging ### ASI04 - Supply Chain Vulnerabilities (NONE) @@ -64,35 +67,34 @@ agsec addresses **7 of 10** OWASP Agentic risks. We are transparent about the 3 **What agsec does:** Not addressed. agsec operates at the action execution layer, not the tool supply chain layer. Tool provenance verification, SBOM analysis, and MCP package integrity checking are not in scope. -**Industry note:** This is an unsolved problem industry-wide. No tool provides comprehensive cryptographic provenance for AI agent tool registries. Tools like Microsoft's Agent Governance Toolkit claim coverage through API gateway allow-listing, but that is policy-level gating, not supply chain verification. - -### ASI05 - Unexpected Code Execution (MEDIUM) +### ASI05 - Unexpected Code Execution (HIGH) **Risk:** Agent generates or runs attacker-controlled code. **What agsec does:** - Gates all bash/shell execution behind policy check -- Regex-based blocking of dangerous patterns (rm, DROP TABLE, etc.) +- Regex-based blocking of dangerous patterns (rm, DROP TABLE, chmod 777, etc.) +- Input normalization catches encoding bypasses (base64 -d | sh, unicode escapes, hex encoding) +- Blocks pipe-to-decoder patterns (`echo ... | base64 -d | sh`) +- Blocks DML/DDL SQL commands - Review effect for human approval before execution - Default policies block destructive commands out of the box -**Limitation:** Cannot sandbox the execution itself. agsec blocks the action before it runs but does not provide process isolation. Pair with OS-level sandboxing (seccomp, Seatbelt) for defense-in-depth. +**Limitation:** Cannot sandbox the execution itself. agsec blocks the action before it runs but does not provide process isolation. + +**agsec features:** 22 threat patterns, input normalization, BlockEncodedExecution, BlockDDL, BlockDML ### ASI06 - Memory & Context Poisoning (NONE) **Risk:** Persistent corruption of RAG stores, embeddings, or agent memory. -**What agsec does:** Not addressed. agsec operates at the action layer, not the memory/context layer. Protecting RAG stores, validating memory writes, and detecting semantic drift are separate concerns. - -**Industry note:** This is largely unsolved. Some tools (Azure Prompt Shields) catch obvious injection in retrieved content, but semantic poisoning that subtly shifts agent behavior is not reliably detectable. +**What agsec does:** Not addressed. agsec operates at the action layer, not the memory/context layer. ### ASI07 - Insecure Inter-Agent Communication (NONE) **Risk:** Spoofed, intercepted, or manipulated agent-to-agent messages. -**What agsec does:** Not addressed. Transport security, message signing, and agent authentication are infrastructure-layer concerns outside agsec's scope. - -**Industry note:** Current approaches rely on TLS + identity providers (e.g., Entra ID), which secure the transport but not the semantic content. Agent-level message validation remains an open research problem. +**What agsec does:** Not addressed. Transport security and agent authentication are infrastructure-layer concerns. ### ASI08 - Cascading Failures (HIGH) @@ -112,10 +114,11 @@ agsec addresses **7 of 10** OWASP Agentic risks. We are transparent about the 3 **What agsec does:** - Review effect forces human approval for sensitive actions -- Immutable audit logs of every action with full context +- Immutable audit logs of every action with full context and actual outcome - Observe mode shows what agents actually do before trusting them +- Threat analysis with consequences helps humans understand real risk -**Limitation:** No behavioral detection or UI safeguards against social engineering. +**agsec features:** Review effect, audit store with outcome logging, `agsec analyze` with blast radius ### ASI10 - Rogue Agents (HIGH) @@ -125,9 +128,13 @@ agsec addresses **7 of 10** OWASP Agentic risks. We are transparent about the 3 - Default-deny bounds what any agent can do, rogue or not - Kill switch (`agsec halt`) for immediate containment - Full audit trail enables forensic investigation -- Policy self-protection (agents can't modify policy files or hook configs) +- Self-protection: agents cannot modify policy files, hook configs, or audit database + - All 6 platform hook configs protected (Claude Code, Codex, Cursor, Windsurf, Cline, Copilot) + - `policies/*.yaml`, `.agsec.yaml` protected from writes + - `audit.db` protected from direct access +- Reads of policy/config files are flagged in threat analysis -**agsec features:** Kill switch, default-deny, audit store, self-protecting policies +**agsec features:** Kill switch, default-deny, audit store, BlockWriteAgsecConfig, BlockAuditDBAccess, read_policy threat pattern ## What's Not Covered @@ -137,4 +144,4 @@ The 3 gaps (ASI04, ASI06, ASI07) require fundamentally different subsystems: - **Memory poisoning** needs memory-layer validation, not action-layer gating - **Inter-agent security** needs transport and semantic message validation -These are industry-wide open problems. We recommend pairing agsec with complementary tools that address these layers (e.g., MCP scanners for supply chain, prompt shields for memory poisoning, mutual TLS for transport security). +These are industry-wide open problems. We recommend pairing agsec with complementary tools that address these layers. diff --git a/docs/policies.md b/docs/policies.md index 9064f06..57a9397 100644 --- a/docs/policies.md +++ b/docs/policies.md @@ -1,6 +1,6 @@ # Policy Format -Policies are YAML files in a `policies/` directory. All files are loaded and merged automatically in alphabetical order. Deny from any file wins. +Policies are YAML files in a `policies/` directory. All files are loaded and merged automatically in alphabetical order. Deny from any file wins — same evaluation logic as AWS IAM. ## Schema @@ -36,6 +36,19 @@ Same as AWS IAM: 3. **Explicit allow** — matching allow rule permits the action 4. **Default policy** — if nothing matches, fall back to default (deny recommended) +## Layered Policy Evaluation + +Policies are evaluated in layers — project policies + optional agent-specific overlay. Each layer is a gate: ALL must allow for an action to proceed. + +``` +Project layer (policies/) → Agent layer (~/.agsec/agents/{name}/) → Result + ↓ ↓ + DENY = blocked DENY = blocked + ALLOW = next layer ALLOW = final allow +``` + +Agent layers can only add restrictions, never widen project permissions. Create agent-specific policies at `~/.agsec/agents/{agent-name}/`. + ## Action Names | Action | What It Covers | @@ -49,6 +62,7 @@ Same as AWS IAM: | `file.glob` | File pattern search | | `file.grep` | Content search | | `agent.spawn` | Sub-agent creation | +| `notebook.edit` | Notebook editing | | `mcp.*` | Any MCP tool calls | | `internal.*` | IDE/platform internals (always allowed) | @@ -84,15 +98,50 @@ conditions: value: ".internal.com" ``` +## Input Normalization + +String values are automatically decoded before policy evaluation to prevent encoding bypasses: + +- **Base64** — pure base64 strings are decoded (e.g., `cm0gLXJmIC8=` → `rm -rf /`) +- **Unicode escapes** — `\u0052` decoded to `R` (catches `D\u0052OP TABLE`) +- **Hex escapes** — `\x72\x6d` decoded to `rm` + +This is non-destructive: only decodes if the result is valid printable text. + ## Default Policies `agsec init` ships with 5 policy files: -- **01_base.yaml** — Default deny. Allow reads, agent spawn, internal tools. -- **02_bash.yaml** — Block `rm`, `DROP TABLE`, secret access, data exfiltration. Allow other bash. -- **03_files.yaml** — Block writes to `.env`, credentials, system dirs, agsec config. Allow other writes. -- **04_web.yaml** — Review external HTTP fetches. Allow localhost and web search. -- **05_git.yaml** — Block force push, protected branches, `reset --hard`. +### 01_base.yaml +- Default deny +- Block reads of secret files (`.env`, credentials, SSH keys, cloud credentials) +- Allow other reads, agent spawn, internal tools + +### 02_bash.yaml +- Block `rm` — file deletion +- Block `chmod 777`, `mkfs`, `dd`, `shred` — destructive filesystem +- Block `DROP TABLE`, `TRUNCATE`, `ALTER DROP` — DDL commands +- Block `DELETE FROM`, `UPDATE SET`, `INSERT INTO` — DML commands +- Block `sqlite3 audit.db` — audit database tampering +- Block `cat .env`, `cat credentials` — secret access via bash +- Block `curl --data .env` — data exfiltration +- Block `base64 -d | sh`, `xxd -r | bash`, `python -c eval | sh` — encoded execution +- Allow other bash + +### 03_files.yaml +- Block writes to `.env`, credentials, SSH keys, cloud credentials +- Block writes to agsec config and hook files (all 6 platforms) +- Block writes to system directories (`/etc`, `/usr`, `/var`, etc.) +- Allow other file writes and notebook edits + +### 04_web.yaml +- Review external HTTP fetches (non-localhost) +- Allow localhost and web search + +### 05_git.yaml +- Block force push (`git push --force`) +- Block push to protected branches (main, master, production, release) +- Block destructive git (`reset --hard`, `clean -f`) ## Multiple Policy Files diff --git a/docs/sdk.md b/docs/sdk.md index e75a7d6..025ee4b 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -79,6 +79,30 @@ print(result.metadata["sid"]) # "BlockFileDelete" print(result.metadata["matched_by"]) # "explicit_deny" ``` +### LayeredPolicyEngine + +Compose multiple policy engines as layers — like AWS IAM permission boundaries. Each layer is a gate: all must allow. + +```python +from agsec.policy import PolicyEngine, LayeredPolicyEngine + +layered = LayeredPolicyEngine() + +# Project policies (default deny) +project = PolicyEngine() +project.load_from_directory("./policies/") +layered.add_layer("project", project) + +# Agent-specific restrictions (default allow — can only add restrictions) +from agsec.types import PolicyStatus +agent = PolicyEngine() +agent.load_from_directory("~/.agsec/agents/my-agent/") +agent._default = PolicyStatus.ALLOW +layered.add_layer("agent", agent) + +result = layered.evaluate("bash.execute", {"command": "ls"}) +``` + ### Validation ```python @@ -100,6 +124,13 @@ with AuditStore("./audit.db") as audit: executions = audit.get_executions(action="payment.charge", limit=50) stats = audit.get_execution_stats() audit.export_to_json("export.json") + + # Time-filtered queries + recent = audit.get_executions_since(hours=24) + + # Retention + deleted = audit.prune(days=7) # delete records older than 7 days + deleted = audit.clear() # delete all records ``` Stats returns: @@ -107,6 +138,11 @@ Stats returns: {"total_executions": 142, "allowed": 100, "blocked": 30, "reviewed": 12, "errors": 0} ``` +Auto-prune via environment variable: +```bash +export AGSEC_AUDIT_RETENTION_DAYS=7 +``` + ## Types ```python diff --git a/examples/openrouter_example.py b/examples/openrouter_example.py index 8b60989..cd6ecd1 100644 --- a/examples/openrouter_example.py +++ b/examples/openrouter_example.py @@ -69,7 +69,7 @@ # Use any model available on OpenRouter response = client.chat.completions.create( - model="meta-llama/llama-3-70b-instruct", + model="minimax/minimax-m2.5:free", messages=[{"role": "user", "content": "Search for the latest AI security news"}], tools=tools, ) diff --git a/examples/openrouter_streaming_example.py b/examples/openrouter_streaming_example.py new file mode 100644 index 0000000..79e3cb0 --- /dev/null +++ b/examples/openrouter_streaming_example.py @@ -0,0 +1,114 @@ +""" +OpenRouter streaming example with agsec protection. + +Demonstrates that streaming works with agsec — chunks flow in real-time, +tool calls are checked after the stream completes. + +Run: pip install agsec[openai] && export OPENROUTER_API_KEY=sk-or-... + python examples/openrouter_streaming_example.py +""" + +import os + +from openai import OpenAI + +from agsec.integrations.openai import protect, allow, deny, param + +client = protect( + OpenAI( + base_url="https://openrouter.ai/api/v1", + api_key=os.environ.get("OPENROUTER_API_KEY"), + ), + allow("search_web", "get_weather"), + deny("execute_code"), + deny("transfer_funds").when(param("amount") > 5000), +) + +tools = [ + { + "type": "function", + "function": { + "name": "search_web", + "description": "Search the web for information", + "parameters": { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "execute_code", + "description": "Execute arbitrary Python code", + "parameters": { + "type": "object", + "properties": {"code": {"type": "string"}}, + "required": ["code"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "transfer_funds", + "description": "Transfer money to a recipient", + "parameters": { + "type": "object", + "properties": { + "to": {"type": "string"}, + "amount": {"type": "number"}, + }, + "required": ["to", "amount"], + }, + }, + }, +] + +print("=== OpenRouter Streaming + agsec ===\n") + +# Stream the response — chunks arrive in real-time +stream = client.chat.completions.create( + model="qwen/qwen3-next-80b-a3b-instruct:free", + messages=[ + { + "role": "system", + "content": "You are a helpful assistant. Always use the provided tools to complete tasks. Do not describe what you would do — call the tools directly.", + }, + { + "role": "user", + "content": "Do all of these right now using the tools: search the web for 'AI agent security', execute code print('hello'), and transfer $10000 to vendor@evil.com", + }, + ], + tools=tools, + stream=True, +) + +# Chunks flow through unchanged — zero latency impact +tool_calls_seen = False +for chunk in stream: + if not chunk.choices: + continue + delta = chunk.choices[0].delta + + # Print streamed text + if delta.content: + print(delta.content, end="", flush=True) + + # Note tool call chunks (just for display) + if getattr(delta, "tool_calls", None): + tool_calls_seen = True + +print() + +# After stream completes, check what agsec caught +if stream._agsec_blocked: + print(f"\nagsec blocked {len(stream._agsec_blocked)} tool call(s):") + for b in stream._agsec_blocked: + print(f" BLOCKED: {b['name']} — {b['reason']} (sid: {b['sid']})") +else: + print("\nagsec: all tool calls allowed") + +if tool_calls_seen and not stream._agsec_blocked: + print("(tool calls were made and all passed policy)") diff --git a/pyproject.toml b/pyproject.toml index aac8767..72dd0a5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "agsec" -version = "0.2.1" +version = "0.2.2" description = "AI Agent Action Firewall core SDK" readme = "README.md" requires-python = ">=3.8" @@ -29,7 +29,8 @@ agsec = "agsec.cli.main:main" openai = ["openai>=1.0.0"] anthropic = ["anthropic>=0.20.0"] langchain = ["langchain-core>=0.2.0"] -all = ["openai>=1.0.0", "anthropic>=0.20.0", "langchain-core>=0.2.0"] +rich = ["rich>=13.0.0"] +all = ["openai>=1.0.0", "anthropic>=0.20.0", "langchain-core>=0.2.0", "rich>=13.0.0"] dev = [ "pytest>=6.0", "pytest-asyncio>=0.21.0", @@ -44,6 +45,9 @@ dev = [ "Repository" = "https://github.com/riyandhiman14/Agent-Sec" "Documentation" = "https://github.com/riyandhiman14/Agent-Sec#readme" +[tool.setuptools.package-data] +agsec = ["templates/policies/*.yaml"] + [tool.pytest.ini_options] minversion = "6.0" addopts = "-ra -q" diff --git a/setup.py b/setup.py index 714a8de..59efdbb 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setup( name="agsec", - version="0.2.1", + version="0.2.2", author="Riyandhiman", author_email="noreply@example.com", description="AI Agent Action Firewall core SDK", @@ -13,6 +13,7 @@ long_description_content_type="text/markdown", url="https://github.com/riyandhiman14/Agent-Sec", packages=find_packages(), + package_data={"agsec": ["templates/policies/*.yaml"]}, license="Apache-2.0", classifiers=[ "Programming Language :: Python :: 3", diff --git a/tests/test_analyze.py b/tests/test_analyze.py index 017aa7f..8cbc59f 100644 --- a/tests/test_analyze.py +++ b/tests/test_analyze.py @@ -37,6 +37,36 @@ def test_destructive_sql_critical(self): assert report.severity_counts["critical"] == 1 assert report.threats[0].pattern.id == "destructive_sql" + def test_dml_sql_high(self): + rows = [_row("bash.execute", {"command": "DELETE FROM users WHERE id=1"})] + report = ThreatClassifier().classify(rows) + assert any(t.pattern.id == "dml_sql" for t in report.threats) + + def test_dml_update_high(self): + rows = [_row("bash.execute", {"command": "UPDATE users SET role='admin'"})] + report = ThreatClassifier().classify(rows) + assert any(t.pattern.id == "dml_sql" for t in report.threats) + + def test_dml_insert_high(self): + rows = [_row("bash.execute", {"command": "INSERT INTO users VALUES (1, 'hacker')"})] + report = ThreatClassifier().classify(rows) + assert any(t.pattern.id == "dml_sql" for t in report.threats) + + def test_audit_tampering_critical(self): + rows = [_row("bash.execute", {"command": "sqlite3 ~/.agsec/audit.db 'DELETE FROM executions'"})] + report = ThreatClassifier().classify(rows) + assert any(t.pattern.id == "audit_tampering" for t in report.threats) + + def test_encoded_execution_critical(self): + rows = [_row("bash.execute", {"command": "echo abc | base64 -d | sh"})] + report = ThreatClassifier().classify(rows) + assert any(t.pattern.id == "encoded_execution" for t in report.threats) + + def test_audit_tampering_psql(self): + rows = [_row("bash.execute", {"command": "psql -d .agsec/audit.db -c 'SELECT *'"})] + report = ThreatClassifier().classify(rows) + assert any(t.pattern.id == "audit_tampering" for t in report.threats) + def test_file_deletion_high(self): rows = [_row("bash.execute", {"command": "rm -rf /tmp/stuff"})] report = ThreatClassifier().classify(rows) @@ -91,6 +121,106 @@ def test_safe_action_no_threat(self): assert len(report.threats) == 0 assert report.blast_radius == 0 + def test_read_secrets_critical(self): + rows = [_row("file.read", {"file_path": ".env"})] + report = ThreatClassifier().classify(rows) + assert report.severity_counts["critical"] == 1 + assert report.threats[0].pattern.id == "read_secrets" + + def test_read_credentials_json_critical(self): + rows = [_row("file.read", {"file_path": "/app/credentials.json"})] + report = ThreatClassifier().classify(rows) + assert report.severity_counts["critical"] == 1 + assert report.threats[0].pattern.id == "read_secrets" + + def test_read_ssh_key_critical(self): + rows = [_row("file.read", {"file_path": "/home/user/.ssh/id_rsa"})] + report = ThreatClassifier().classify(rows) + assert report.severity_counts["critical"] == 1 + assert report.threats[0].pattern.id == "read_secrets" + + def test_read_system_files_high(self): + rows = [_row("file.read", {"file_path": "/etc/passwd"})] + report = ThreatClassifier().classify(rows) + assert report.severity_counts["high"] == 1 + assert report.threats[0].pattern.id == "read_system" + + def test_read_policy_config_high(self): + rows = [_row("file.read", {"file_path": ".agsec.yaml"})] + report = ThreatClassifier().classify(rows) + assert report.severity_counts["high"] == 1 + assert report.threats[0].pattern.id == "read_policy" + + def test_read_claude_settings_high(self): + rows = [_row("file.read", {"file_path": ".claude/settings.json"})] + report = ThreatClassifier().classify(rows) + assert report.severity_counts["high"] == 1 + assert report.threats[0].pattern.id == "read_policy" + + def test_policy_tampering_cursor(self): + rows = [_row("file.edit", {"file_path": ".cursor/hooks.json"})] + report = ThreatClassifier().classify(rows) + assert any(t.pattern.id == "policy_tampering" for t in report.threats) + + def test_policy_tampering_windsurf(self): + rows = [_row("file.edit", {"file_path": ".windsurf/settings.json"})] + report = ThreatClassifier().classify(rows) + assert any(t.pattern.id == "policy_tampering" for t in report.threats) + + def test_notebook_sensitive_high(self): + rows = [_row("notebook.edit", {"file_path": ".env"})] + report = ThreatClassifier().classify(rows) + assert report.severity_counts["high"] == 1 + assert report.threats[0].pattern.id == "notebook_sensitive" + + def test_scan_secrets_via_glob_medium(self): + rows = [_row("file.glob", {"pattern": "**/.env*"})] + report = ThreatClassifier().classify(rows) + assert report.severity_counts["medium"] == 1 + assert report.threats[0].pattern.id == "scan_secrets" + + def test_scan_secrets_via_grep_medium(self): + rows = [_row("file.grep", {"pattern": "password"})] + report = ThreatClassifier().classify(rows) + assert report.severity_counts["medium"] == 1 + assert report.threats[0].pattern.id == "scan_secrets" + + +class TestOutcomeClassification: + def test_observe_mode_allowed_through_is_threat(self): + """In observe mode, outcome=allowed even though policy said block — it's a threat.""" + row = _row("bash.execute", {"command": "cat .env"}, policy_status="block") + row["outcome"] = "allowed" + report = ThreatClassifier().classify([row]) + assert len(report.threats) == 1 + assert len(report.blocked) == 0 + assert report.blast_radius > 0 + + def test_enforce_mode_blocked_is_caught(self): + row = _row("bash.execute", {"command": "cat .env"}, policy_status="block") + row["outcome"] = "blocked" + report = ThreatClassifier().classify([row]) + assert len(report.threats) == 0 + assert len(report.blocked) == 1 + + def test_mixed_outcomes(self): + rows = [ + {**_row("bash.execute", {"command": "cat .env"}, policy_status="block"), "outcome": "blocked"}, + {**_row("bash.execute", {"command": "rm -rf /tmp"}, policy_status="block"), "outcome": "allowed"}, + {**_row("file.read", {"file_path": ".env"}, policy_status="block"), "outcome": "allowed"}, + ] + report = ThreatClassifier().classify(rows) + assert len(report.blocked) == 1 # cat .env was actually blocked + assert len(report.threats) == 2 # rm + .env read got through + + def test_legacy_rows_without_outcome_use_policy_status(self): + """Old audit rows without outcome column fall back to policy_status.""" + row = _row("bash.execute", {"command": "cat .env"}, policy_status="block") + # No outcome field — legacy behavior + report = ThreatClassifier().classify([row]) + assert len(report.blocked) == 1 + assert len(report.threats) == 0 + class TestBlockedVsAllowed: def test_blocked_goes_to_blocked_list(self): @@ -229,6 +359,71 @@ def test_sorted_by_severity(self): assert severities == ["critical", "medium", "low"] +class TestActivityGroups: + """Tests for --all activity report grouping.""" + + def test_groups_by_action_type(self): + from agsec.cli.commands.analyze import _build_activity_groups + rows = [ + _row("bash.execute", {"command": "ls"}), + _row("bash.execute", {"command": "pwd"}), + _row("file.read", {"file_path": "README.md"}), + ] + groups = _build_activity_groups(rows) + assert "bash.execute" in groups + assert "file.read" in groups + assert groups["bash.execute"]["label"] == "Shell Commands" + assert groups["file.read"]["label"] == "File Reads" + assert sum(groups["bash.execute"]["counts"].values()) == 2 + assert sum(groups["file.read"]["counts"].values()) == 1 + + def test_blocked_items_tracked(self): + from agsec.cli.commands.analyze import _build_activity_groups + rows = [ + _row("bash.execute", {"command": "rm -rf /"}, policy_status="block"), + _row("bash.execute", {"command": "ls"}, policy_status="allow"), + ] + groups = _build_activity_groups(rows) + assert groups["bash.execute"]["counts"]["block"] == 1 + assert groups["bash.execute"]["counts"]["allow"] == 1 + + def test_internal_tools_skipped(self): + from agsec.cli.commands.analyze import _build_activity_groups + rows = [ + _row("internal.TaskCreate", {"name": "test"}), + _row("bash.execute", {"command": "ls"}), + ] + groups = _build_activity_groups(rows) + assert "internal.TaskCreate" not in groups + assert "bash.execute" in groups + + def test_mcp_tools_labeled(self): + from agsec.cli.commands.analyze import _build_activity_groups + rows = [_row("mcp.github.create_issue", {"title": "bug"})] + groups = _build_activity_groups(rows) + assert "mcp.github.create_issue" in groups + assert groups["mcp.github.create_issue"]["label"] == "MCP: github.create_issue" + + def test_extracts_display_value(self): + from agsec.cli.commands.analyze import _build_activity_groups + rows = [_row("bash.execute", {"command": "npm install"})] + groups = _build_activity_groups(rows) + assert groups["bash.execute"]["items"][0]["value"] == "npm install" + + def test_truncates_long_values(self): + from agsec.cli.commands.analyze import _build_activity_groups + long_cmd = "x" * 200 + rows = [_row("bash.execute", {"command": long_cmd})] + groups = _build_activity_groups(rows) + assert len(groups["bash.execute"]["items"][0]["value"]) == 100 + + def test_handles_string_params(self): + from agsec.cli.commands.analyze import _build_activity_groups + rows = [{"action": "bash.execute", "params": '{"command": "ls"}', "policy_status": "allow"}] + groups = _build_activity_groups(rows) + assert groups["bash.execute"]["items"][0]["value"] == "ls" + + class TestAuditStoreTimeQuery: def test_get_executions_since_returns_all_when_no_filter(self): store = AuditStore(":memory:") @@ -258,3 +453,91 @@ def test_get_executions_since_filters_by_hours(self): results = store.get_executions_since(hours=24) assert len(results) == 1 assert "ls" in results[0]["params"] + + +class TestAuditRetention: + def test_prune_deletes_old_records(self): + from datetime import datetime, timedelta + store = AuditStore(":memory:") + recent = datetime.utcnow().isoformat() + old = (datetime.utcnow() - timedelta(hours=48)).isoformat() + store.conn.execute( + "INSERT INTO executions (timestamp, action, params, policy_status) VALUES (?, ?, ?, ?)", + (recent, "bash.execute", '{"command": "ls"}', "allow"), + ) + store.conn.execute( + "INSERT INTO executions (timestamp, action, params, policy_status) VALUES (?, ?, ?, ?)", + (old, "bash.execute", '{"command": "old"}', "allow"), + ) + store.conn.commit() + store.prune(days=1) + results = store.get_executions() + assert len(results) == 1 + assert "ls" in results[0]["params"] + + def test_prune_returns_count(self): + from datetime import datetime, timedelta + store = AuditStore(":memory:") + old = (datetime.utcnow() - timedelta(days=10)).isoformat() + for i in range(5): + store.conn.execute( + "INSERT INTO executions (timestamp, action, params, policy_status) VALUES (?, ?, ?, ?)", + (old, "bash.execute", f'{{"command": "cmd{i}"}}', "allow"), + ) + store.conn.commit() + count = store.prune(days=7) + assert count == 5 + + def test_clear_deletes_all(self): + store = AuditStore(":memory:") + for i in range(3): + store.conn.execute( + "INSERT INTO executions (timestamp, action, params, policy_status) VALUES (?, ?, ?, ?)", + ("2026-03-28T12:00:00", "bash.execute", f'{{"command": "cmd{i}"}}', "allow"), + ) + store.conn.commit() + count = store.clear() + assert count == 3 + results = store.get_executions() + assert len(results) == 0 + + def test_auto_prune_with_env_var(self, monkeypatch): + from datetime import datetime, timedelta + old = (datetime.utcnow() - timedelta(days=10)).isoformat() + + # Create a store and insert old data directly + import sqlite3 + conn = sqlite3.connect(":memory:") + conn.execute(""" + CREATE TABLE executions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp TEXT NOT NULL, + action TEXT NOT NULL, + params TEXT NOT NULL, + result TEXT, + policy_status TEXT NOT NULL, + policy_reason TEXT, + context TEXT, + error TEXT, + outcome TEXT + ) + """) + conn.execute( + "INSERT INTO executions (timestamp, action, params, policy_status) VALUES (?, ?, ?, ?)", + (old, "bash.execute", '{"command": "old"}', "allow"), + ) + conn.commit() + + # Now create AuditStore with env var — but we can't reuse the connection + # Instead, test prune directly with env var + monkeypatch.setenv("AGSEC_AUDIT_RETENTION_DAYS", "7") + store = AuditStore(":memory:") + store.conn.execute( + "INSERT INTO executions (timestamp, action, params, policy_status) VALUES (?, ?, ?, ?)", + (old, "bash.execute", '{"command": "old"}', "allow"), + ) + store.conn.commit() + # Auto-prune already ran on init but the insert was after — prune manually + store._auto_prune() + results = store.get_executions() + assert len(results) == 0 diff --git a/tests/test_default_policies.py b/tests/test_default_policies.py index 4791c38..ba893fa 100644 --- a/tests/test_default_policies.py +++ b/tests/test_default_policies.py @@ -46,6 +46,37 @@ def test_block_rm_force(self, engine): def test_block_drop_table(self, engine): result = engine.evaluate("bash.execute", {"command": "psql -c 'DROP TABLE users'"}) assert result.status == PolicyStatus.BLOCK + assert result.metadata["sid"] == "BlockDDL" + + def test_block_truncate_table(self, engine): + result = engine.evaluate("bash.execute", {"command": "TRUNCATE TABLE sessions"}) + assert result.status == PolicyStatus.BLOCK + + def test_block_alter_drop(self, engine): + result = engine.evaluate("bash.execute", {"command": "ALTER TABLE users DROP column email"}) + assert result.status == PolicyStatus.BLOCK + + def test_block_delete_from(self, engine): + result = engine.evaluate("bash.execute", {"command": "DELETE FROM users WHERE id=1"}) + assert result.status == PolicyStatus.BLOCK + assert result.metadata["sid"] == "BlockDML" + + def test_block_update_set(self, engine): + result = engine.evaluate("bash.execute", {"command": "UPDATE users SET role='admin'"}) + assert result.status == PolicyStatus.BLOCK + + def test_block_insert_into(self, engine): + result = engine.evaluate("bash.execute", {"command": "INSERT INTO users VALUES (1, 'hacker')"}) + assert result.status == PolicyStatus.BLOCK + + def test_block_audit_db_access(self, engine): + result = engine.evaluate("bash.execute", {"command": "sqlite3 ~/.agsec/audit.db 'SELECT * FROM executions'"}) + assert result.status == PolicyStatus.BLOCK + assert result.metadata["sid"] == "BlockAuditDBAccess" + + def test_block_audit_db_psql(self, engine): + result = engine.evaluate("bash.execute", {"command": "psql -d .agsec/audit.db"}) + assert result.status == PolicyStatus.BLOCK def test_block_cat_env(self, engine): result = engine.evaluate("bash.execute", {"command": "cat .env"}) @@ -67,6 +98,27 @@ def test_allow_git_push_feature(self, engine): result = engine.evaluate("bash.execute", {"command": "git push origin feature/my-branch"}) assert result.status == PolicyStatus.ALLOW + def test_block_base64_pipe_to_sh(self, engine): + result = engine.evaluate("bash.execute", + {"command": 'echo "cm0gLXJmIC8=" | base64 -d | sh'}) + assert result.status == PolicyStatus.BLOCK + assert result.metadata["sid"] == "BlockEncodedExecution" + + def test_block_xxd_pipe_to_bash(self, engine): + result = engine.evaluate("bash.execute", + {"command": 'echo 726d202d7266 | xxd -r -p | bash'}) + assert result.status == PolicyStatus.BLOCK + + def test_block_python_eval_pipe(self, engine): + result = engine.evaluate("bash.execute", + {"command": "python3 -c 'exec(\"import os\")' | sh"}) + assert result.status == PolicyStatus.BLOCK + + def test_allow_normal_base64_command(self, engine): + result = engine.evaluate("bash.execute", + {"command": "echo 'hello' | base64"}) + assert result.status == PolicyStatus.ALLOW + class TestFilePolicies: def test_allow_write_source(self, engine): @@ -93,12 +145,53 @@ def test_block_edit_env(self, engine): result = engine.evaluate("file.edit", {"file_path": ".env.production"}) assert result.status == PolicyStatus.BLOCK + def test_block_write_cursor_hooks(self, engine): + result = engine.evaluate("file.write", {"file_path": ".cursor/hooks.json", "content": "{}"}) + assert result.status == PolicyStatus.BLOCK + + def test_block_write_windsurf_settings(self, engine): + result = engine.evaluate("file.write", {"file_path": ".windsurf/settings.json", "content": "{}"}) + assert result.status == PolicyStatus.BLOCK + + def test_block_write_cline_hooks(self, engine): + result = engine.evaluate("file.write", {"file_path": ".clinerules/hooks/agsec-check.sh", "content": "#!/bin/bash"}) + assert result.status == PolicyStatus.BLOCK + + def test_block_write_copilot_hooks(self, engine): + result = engine.evaluate("file.write", {"file_path": ".github/hooks/hooks.json", "content": "{}"}) + assert result.status == PolicyStatus.BLOCK + class TestReadPolicies: def test_allow_read_any(self, engine): result = engine.evaluate("file.read", {"file_path": "anything.py"}) assert result.status == PolicyStatus.ALLOW + def test_block_read_env(self, engine): + result = engine.evaluate("file.read", {"file_path": ".env"}) + assert result.status == PolicyStatus.BLOCK + assert result.metadata["sid"] == "BlockReadSecrets" + + def test_block_read_credentials_json(self, engine): + result = engine.evaluate("file.read", {"file_path": "/app/credentials.json"}) + assert result.status == PolicyStatus.BLOCK + + def test_block_read_ssh_key(self, engine): + result = engine.evaluate("file.read", {"file_path": "/home/user/.ssh/id_rsa"}) + assert result.status == PolicyStatus.BLOCK + + def test_block_read_aws_credentials(self, engine): + result = engine.evaluate("file.read", {"file_path": "/home/user/.aws/credentials"}) + assert result.status == PolicyStatus.BLOCK + + def test_block_read_env_local(self, engine): + result = engine.evaluate("file.read", {"file_path": ".env.local"}) + assert result.status == PolicyStatus.BLOCK + + def test_allow_read_normal_file(self, engine): + result = engine.evaluate("file.read", {"file_path": "src/main.py"}) + assert result.status == PolicyStatus.ALLOW + def test_allow_glob(self, engine): result = engine.evaluate("file.glob", {"pattern": "**/*.py"}) assert result.status == PolicyStatus.ALLOW diff --git a/tests/test_integrations_anthropic.py b/tests/test_integrations_anthropic.py index bd4ab9e..94e653b 100644 --- a/tests/test_integrations_anthropic.py +++ b/tests/test_integrations_anthropic.py @@ -167,3 +167,113 @@ def test_review_treated_as_block(self): result = protected.messages.create(model="claude-sonnet-4-20250514", messages=[]) assert len(result.content) == 0 assert result._agsec_blocked[0]["status"] == "review" + + +# --------------------------------------------------------------------------- +# Streaming tests +# --------------------------------------------------------------------------- + + +def _make_stream_event(event_type, **kwargs): + """Create a mock Anthropic stream event.""" + event = SimpleNamespace() + event.type = event_type + for k, v in kwargs.items(): + setattr(event, k, v) + return event + + +def _make_content_block_start(index, name, block_id="tu_1"): + cb = SimpleNamespace() + cb.type = "tool_use" + cb.id = block_id + cb.name = name + cb.input = {} + return _make_stream_event("content_block_start", index=index, content_block=cb) + + +def _make_input_json_delta(index, partial_json): + delta = SimpleNamespace() + delta.type = "input_json_delta" + delta.partial_json = partial_json + return _make_stream_event("content_block_delta", index=index, delta=delta) + + +def _make_content_block_stop(index): + return _make_stream_event("content_block_stop", index=index) + + +def _make_stream_client_anthropic(events): + client = MagicMock() + client.messages.create = MagicMock(return_value=iter(events)) + return client + + +class TestStreaming: + def test_streaming_returns_guarded_stream(self): + events = [_make_stream_event("message_start")] + client = _make_stream_client_anthropic(events) + protected = protect(client, allow("search")) + + result = protected.messages.create(model="claude-sonnet-4-20250514", messages=[], stream=True) + assert hasattr(result, "_agsec_blocked") + + def test_streaming_events_yielded_unchanged(self): + events = [ + _make_stream_event("message_start"), + _make_stream_event("message_stop"), + ] + client = _make_stream_client_anthropic(events) + protected = protect(client, allow("search")) + + stream = protected.messages.create(model="claude-sonnet-4-20250514", messages=[], stream=True) + received = list(stream) + assert len(received) == 2 + assert received[0].type == "message_start" + assert received[1].type == "message_stop" + + def test_streaming_allowed_tool_not_blocked(self): + events = [ + _make_content_block_start(0, "search", "tu_1"), + _make_input_json_delta(0, '{"query": "test"}'), + _make_content_block_stop(0), + ] + client = _make_stream_client_anthropic(events) + protected = protect(client, allow("search")) + + stream = protected.messages.create(model="claude-sonnet-4-20250514", messages=[], stream=True) + list(stream) + assert stream._agsec_blocked == [] + + def test_streaming_blocked_tool_detected(self): + events = [ + _make_content_block_start(0, "delete_user", "tu_1"), + _make_input_json_delta(0, '{"id": '), + _make_input_json_delta(0, '123}'), + _make_content_block_stop(0), + ] + client = _make_stream_client_anthropic(events) + protected = protect(client, deny("delete_user")) + + stream = protected.messages.create(model="claude-sonnet-4-20250514", messages=[], stream=True) + list(stream) + assert len(stream._agsec_blocked) == 1 + assert stream._agsec_blocked[0]["name"] == "delete_user" + assert stream._agsec_blocked[0]["tool_use_id"] == "tu_1" + + def test_streaming_mixed_tools(self): + events = [ + _make_content_block_start(0, "search", "tu_1"), + _make_input_json_delta(0, '{"query": "test"}'), + _make_content_block_stop(0), + _make_content_block_start(1, "delete_user", "tu_2"), + _make_input_json_delta(1, '{"id": 1}'), + _make_content_block_stop(1), + ] + client = _make_stream_client_anthropic(events) + protected = protect(client, allow("search"), deny("delete_user")) + + stream = protected.messages.create(model="claude-sonnet-4-20250514", messages=[], stream=True) + list(stream) + assert len(stream._agsec_blocked) == 1 + assert stream._agsec_blocked[0]["name"] == "delete_user" diff --git a/tests/test_integrations_openai.py b/tests/test_integrations_openai.py index ddf71f2..d742bbd 100644 --- a/tests/test_integrations_openai.py +++ b/tests/test_integrations_openai.py @@ -178,3 +178,112 @@ def test_review_treated_as_block(self): result = protected.chat.completions.create(model="gpt-4", messages=[]) assert result.choices[0].message.tool_calls is None assert result._agsec_blocked[0]["status"] == "review" + + +# --------------------------------------------------------------------------- +# Streaming tests +# --------------------------------------------------------------------------- + + +def _make_stream_chunk(tool_calls=None, finish_reason=None, content=None): + """Create a mock OpenAI streaming chunk.""" + delta = SimpleNamespace() + delta.tool_calls = tool_calls + delta.content = content + delta.role = None + + choice = SimpleNamespace() + choice.delta = delta + choice.finish_reason = finish_reason + choice.index = 0 + + chunk = SimpleNamespace() + chunk.choices = [choice] + chunk.id = "chatcmpl-stream" + return chunk + + +def _make_tool_call_delta(index, tc_id=None, name=None, arguments=None): + """Create a mock tool call delta for streaming.""" + tc = SimpleNamespace() + tc.index = index + tc.id = tc_id + tc.function = SimpleNamespace() + tc.function.name = name + tc.function.arguments = arguments + return tc + + +def _make_stream_client(chunks): + """Create a mock client that returns an iterable stream.""" + client = MagicMock() + client.chat.completions.create = MagicMock(return_value=iter(chunks)) + return client + + +class TestStreaming: + def test_streaming_returns_guarded_stream(self): + chunks = [_make_stream_chunk(content="hello")] + client = _make_stream_client(chunks) + protected = protect(client, allow("search")) + + result = protected.chat.completions.create(model="gpt-4", messages=[], stream=True) + # Should not raise NotImplementedError + assert hasattr(result, "_agsec_blocked") + + def test_streaming_chunks_yielded_unchanged(self): + chunks = [ + _make_stream_chunk(content="hello"), + _make_stream_chunk(content=" world"), + ] + client = _make_stream_client(chunks) + protected = protect(client, allow("search")) + + stream = protected.chat.completions.create(model="gpt-4", messages=[], stream=True) + received = list(stream) + assert len(received) == 2 + assert received[0].choices[0].delta.content == "hello" + assert received[1].choices[0].delta.content == " world" + + def test_streaming_allowed_tool_not_blocked(self): + chunks = [ + _make_stream_chunk(tool_calls=[_make_tool_call_delta(0, tc_id="tc_1", name="search", arguments='{"qu')]), + _make_stream_chunk(tool_calls=[_make_tool_call_delta(0, arguments='ery": "test"}')]), + _make_stream_chunk(finish_reason="tool_calls"), + ] + client = _make_stream_client(chunks) + protected = protect(client, allow("search")) + + stream = protected.chat.completions.create(model="gpt-4", messages=[], stream=True) + received = list(stream) + assert len(received) == 3 + assert stream._agsec_blocked == [] + + def test_streaming_blocked_tool_detected(self): + chunks = [ + _make_stream_chunk(tool_calls=[_make_tool_call_delta(0, tc_id="tc_1", name="delete_user", arguments='{"id')]), + _make_stream_chunk(tool_calls=[_make_tool_call_delta(0, arguments='": 123}')]), + _make_stream_chunk(finish_reason="tool_calls"), + ] + client = _make_stream_client(chunks) + protected = protect(client, deny("delete_user")) + + stream = protected.chat.completions.create(model="gpt-4", messages=[], stream=True) + list(stream) # consume + assert len(stream._agsec_blocked) == 1 + assert stream._agsec_blocked[0]["name"] == "delete_user" + assert stream._agsec_blocked[0]["tool_call_id"] == "tc_1" + + def test_streaming_mixed_tools(self): + chunks = [ + _make_stream_chunk(tool_calls=[_make_tool_call_delta(0, tc_id="tc_1", name="search", arguments='{"query": "test"}')]), + _make_stream_chunk(tool_calls=[_make_tool_call_delta(1, tc_id="tc_2", name="delete_user", arguments='{"id": 1}')]), + _make_stream_chunk(finish_reason="tool_calls"), + ] + client = _make_stream_client(chunks) + protected = protect(client, allow("search"), deny("delete_user")) + + stream = protected.chat.completions.create(model="gpt-4", messages=[], stream=True) + list(stream) + assert len(stream._agsec_blocked) == 1 + assert stream._agsec_blocked[0]["name"] == "delete_user" diff --git a/tests/test_output.py b/tests/test_output.py new file mode 100644 index 0000000..07234b5 --- /dev/null +++ b/tests/test_output.py @@ -0,0 +1,60 @@ +"""Tests for CLI output utilities.""" + +import io +import sys + +from agsec.cli.output import ( + RICH, + mode_label, + severity_label, + table, +) + + +class TestOutputFallback: + def test_mode_label_enforce(self): + label = mode_label("enforce") + assert "ENFORCE" in label + + def test_mode_label_observe(self): + label = mode_label("observe") + assert "OBSERVE" in label + + def test_mode_label_halt(self): + label = mode_label("halt") + assert "HALT" in label + + def test_severity_label_critical(self): + label = severity_label("critical") + assert "CRITICAL" in label + + def test_severity_label_low(self): + label = severity_label("low") + assert "LOW" in label + + +class TestTableFormatting: + def test_table_short_row(self, capsys): + """Table should not crash when a row has fewer columns than headers.""" + if RICH: + return # Rich handles this internally + table("Test", ["Col1", "Col2", "Col3"], [["a", "b"]]) + captured = capsys.readouterr() + assert "Col1" in captured.out + assert "a" in captured.out + + def test_table_none_values(self, capsys): + """Table should handle None values in cells.""" + if RICH: + return + table("Test", ["Name", "Value"], [["key", None]]) + captured = capsys.readouterr() + assert "None" in captured.out + + def test_table_empty_rows(self, capsys): + """Table with no rows should not crash.""" + if RICH: + return + table("Test", ["Col1", "Col2"], []) + captured = capsys.readouterr() + assert "Test" in captured.out diff --git a/tests/test_policy_iam.py b/tests/test_policy_iam.py index ad0534d..9b401fe 100644 --- a/tests/test_policy_iam.py +++ b/tests/test_policy_iam.py @@ -903,3 +903,67 @@ def test_metadata_on_default(self, tmp_path): engine.load_from_file(path) result = engine.evaluate("anything", {}) assert result.metadata["matched_by"] == "default" + + +class TestInputNormalization: + """Test that encoding bypasses are caught via input normalization.""" + + def _make_engine(self, tmp_path): + path = _write_policy(tmp_path, "policy.yaml", """ +version: "1.0" +default: allow +statements: + - sid: "BlockRM" + effect: deny + actions: ["bash.execute"] + conditions: + params.command: + op: "regex" + value: "\\\\brm\\\\s" + reason: "rm blocked" + - sid: "BlockDrop" + effect: deny + actions: ["bash.execute"] + conditions: + params.command: + op: "regex" + value: "(?i)DROP\\\\s+TABLE" + reason: "DROP TABLE blocked" +""") + engine = PolicyEngine() + engine.load_from_file(path) + return engine + + def test_base64_decoded_and_matched(self, tmp_path): + engine = self._make_engine(tmp_path) + # "rm -rf /" base64 = "cm0gLXJmIC8=" + result = engine.evaluate("bash.execute", {"command": "cm0gLXJmIC8="}) + assert result.status == PolicyStatus.BLOCK + + def test_unicode_escape_decoded(self, tmp_path): + engine = self._make_engine(tmp_path) + # D\u0052OP TABLE -> DROP TABLE + result = engine.evaluate("bash.execute", {"command": "D\\u0052OP TABLE users"}) + assert result.status == PolicyStatus.BLOCK + + def test_hex_escape_decoded(self, tmp_path): + engine = self._make_engine(tmp_path) + # \x72\x6d = rm + result = engine.evaluate("bash.execute", {"command": "\\x72\\x6d -rf /"}) + assert result.status == PolicyStatus.BLOCK + + def test_normal_string_unchanged(self, tmp_path): + engine = self._make_engine(tmp_path) + result = engine.evaluate("bash.execute", {"command": "ls -la"}) + assert result.status == PolicyStatus.ALLOW + + def test_non_base64_not_decoded(self, tmp_path): + engine = self._make_engine(tmp_path) + # "hello world" has spaces, not treated as base64 + result = engine.evaluate("bash.execute", {"command": "hello world"}) + assert result.status == PolicyStatus.ALLOW + + def test_normal_rm_still_blocked(self, tmp_path): + engine = self._make_engine(tmp_path) + result = engine.evaluate("bash.execute", {"command": "rm -rf /tmp"}) + assert result.status == PolicyStatus.BLOCK diff --git a/tests/test_status.py b/tests/test_status.py new file mode 100644 index 0000000..4c4e249 --- /dev/null +++ b/tests/test_status.py @@ -0,0 +1,54 @@ +"""Tests for agsec status command.""" + +import json +import subprocess +import sys + +import pytest + + +def _run_status(env_overrides=None): + """Run agsec status as subprocess.""" + env = {"AGSEC_MODE": "enforce", "AGSEC_AUDIT_DB": ":memory:"} + if env_overrides: + env.update(env_overrides) + result = subprocess.run( + [sys.executable, "-m", "agsec.cli.main", "status"], + capture_output=True, + text=True, + timeout=10, + env=env, + ) + return result.returncode, result.stdout.strip(), result.stderr.strip() + + +class TestStatusCommand: + def test_status_runs(self): + code, stdout, stderr = _run_status() + assert code == 0 + assert "agsec" in stdout.lower() + + def test_status_shows_mode(self): + code, stdout, _ = _run_status({"AGSEC_MODE": "enforce"}) + assert "ENFORCE" in stdout + + def test_status_shows_observe(self): + code, stdout, _ = _run_status({"AGSEC_MODE": "observe"}) + assert "OBSERVE" in stdout + + def test_status_shows_version(self): + code, stdout, _ = _run_status() + assert "v0." in stdout # matches v0.2.1 etc + + +class TestVersionFlag: + def test_version(self): + result = subprocess.run( + [sys.executable, "-m", "agsec.cli.main", "--version"], + capture_output=True, + text=True, + timeout=10, + ) + assert result.returncode == 0 + assert "agsec" in result.stdout + assert "0.2.2" in result.stdout