Ai Policy As Code: OPA Checks Before Every Shell Execution

{
  "argv": ["curl", "-s", "http://169.254.169.254/latest/meta-data/"],
  "cwd": "/workspace/app",
  "actor": "agent:pr-4421",
  "elevation": false,
  "network": true
}

package agent.shell

default allow := false

dangerous_host if {
  some i
  host := input.argv[i]
  host == "169.254.169.254"
}

dangerous_host if {
  some i
  startswith(input.argv[i], "http://169.254.169.254")
}

deny[msg] if {
  dangerous_host
  msg := "metadata_ip_blocked"
}

deny[msg] if {
  input.argv[0] == "kubectl"
  some i
  input.argv[i] == "delete"
  not input.elevation
  msg := "kubectl_delete_needs_elevation"
}

deny[msg] if {
  input.argv[0] == "aws"
  input.argv[1] == "s3"
  input.argv[2] == "sync"
  not startswith(input.argv[3], "s3://my-org-scratch/")
  msg := "s3_sync_unscoped"
}

allow if {
  count(deny) == 0
  input.argv[0] in {"pnpm", "npm", "pytest", "go", "cargo", "git"}
}

// beforeShell(cmd)
const decision = await opa.evaluate("agent/shell/allow", {
  argv: parse(cmd),
  cwd: process.cwd(),
  actor: session.actor,
  elevation: session.elevated,
});
if (!decision.result) {
  throw new Error(`opa_deny:${decision.denyReasons.join(",")}`);
}

opa test ./policy/agent -v
conftest verify --policy policy/agent fixtures/shell/*.json

type ShellDecision = {
  allow: boolean;
  reasons: string[];
  policySha: string;
  argvHash: string;
  actor: string;
  ticketId?: string;
};

// Persist deny and allow — allows matter for forensics too
await logDecision(decision);