diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..f5428c1 --- /dev/null +++ b/.env.example @@ -0,0 +1,46 @@ +# Copy to .env and fill in. Do NOT commit real secrets. + +# --- Agent behavior --- +SUPERUSER_HUMAN_OWNER=you@example.com +SUPERUSER_APPROVAL_AT=MEDIUM +SUPERUSER_AUDIT_LOG=superuser-audit.log +SUPERUSER_COMMAND_TIMEOUT=60 +# SUPERUSER_CONFIG=superuser.yaml + +# --- Cisco AI Defense --- +AI_DEFENSE_ENABLED=true +AI_DEFENSE_MODE=enforce +# Inspection API key (explicit MCP tool-call inspection): +AI_DEFENSE_INSPECTION_API_KEY= +# Agent Runtime (agentsec.protect) API mode: +AI_DEFENSE_API_MODE_LLM_ENDPOINT= +AI_DEFENSE_API_MODE_LLM_API_KEY= +# Or point at an agentsec.yaml instead of the two vars above: +# AI_DEFENSE_AGENTSEC_CONFIG=agentsec.yaml + +# --- Cisco Duo Agentic Identity (per-tool-call authorization) --- +DUO_ENABLED=false +DUO_MODE=enforce +DUO_AGENT_ID=superuser-agent-1 +# OAuth 2.1 client-credentials (machine-to-machine) endpoint + creds: +DUO_TOKEN_URL= +DUO_CLIENT_ID= +DUO_CLIENT_SECRET= +# Space-separated scopes requested at token time: +DUO_SCOPES=superuser.read superuser.exec +# RFC 8707 resource indicator (audience this agent may access): +DUO_RESOURCE= +# Optional explicit per-tool-call decision endpoint (else scope-based): +DUO_AUTHORIZATION_URL= + +# --- Stripe usage-based (metered) billing --- +STRIPE_ENABLED=false +STRIPE_API_KEY= +STRIPE_CUSTOMER_ID= +STRIPE_METER_EVENT_NAME=superuser_tool_call +STRIPE_DEFAULT_VALUE=1 + +# --- GitHub (least-privilege, fine-grained token) --- +GITHUB_TOKEN= +GITHUB_ALLOWED_REPOS=your-org/your-repo +GITHUB_ALLOW_WRITES=false diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..6f24044 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,18 @@ +name: CI +on: + push: + branches: [main] + pull_request: +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10", "3.12"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - run: pip install -e '.[dev]' + - run: pytest -q diff --git a/.github/workflows/merge-sync-audit.yml b/.github/workflows/merge-sync-audit.yml deleted file mode 100644 index d97f5f0..0000000 --- a/.github/workflows/merge-sync-audit.yml +++ /dev/null @@ -1,76 +0,0 @@ -name: Accepted PR → Inventory Sync + Audit + Site Redeploy Check - -on: - pull_request: - types: [closed] - -jobs: - sync-on-merge: - # Only run for successful merges to main - if: github.event.pull_request.merged == true && github.event.pull_request.base.ref == 'main' - runs-on: ubuntu-latest - permissions: - contents: write - steps: - - name: Checkout repository - uses: actions/checkout@v4 - with: - token: ${{ secrets.GITHUB_TOKEN }} - fetch-depth: 0 - - - name: Configure git - run: | - git config user.name "superuser-guard-bot" - git config user.email "guard-bot@fitzyracing1.github.io" - - - name: Update central inventory, changelog, and append-only audit - env: - PR_NUMBER: ${{ github.event.pull_request.number }} - PR_TITLE: ${{ github.event.pull_request.title }} - PR_AUTHOR: ${{ github.event.pull_request.user.login }} - PR_URL: ${{ github.event.pull_request.html_url }} - run: | - set -e - MERGE_DATE=$(date -u +"%Y-%m-%d %H:%M UTC") - TIMESTAMP=$(date -u +"%Y-%m-%dT%H:%M:%SZ") - - # Structured entry for inventory and changelog - ENTRY="## Merged PR #${PR_NUMBER} — ${PR_TITLE} - -- **Author**: @${PR_AUTHOR} -- **Merged**: ${MERGE_DATE} -- **URL**: ${PR_URL} -- **Action**: inventory + changelog + audit updated | site redeploy triggered - -" - - echo -e " -${ENTRY}" >> projects-inventory.md - echo -e " -${ENTRY}" >> CHANGELOG.md - - # Guard-style append-only audit entry (exactly like superuser guard) - AUDIT_ENTRY="[${TIMESTAMP}] MERGE_DECISION: PR #${PR_NUMBER} \"${PR_TITLE}\" by @${PR_AUTHOR} → main | updates: projects-inventory.md + CHANGELOG.md + AUDIT.md | site_redeploy_check: passed | outcome: success | extends: Rule 2 | guard: one-look-two-fight-three-listen-four-break" - echo "${AUDIT_ENTRY}" >> AUDIT.md - - echo "[merge-sync] Updated inventory, changelog, and audit for PR #${PR_NUMBER}" - - - name: Site redeploy checks - run: | - echo "=== Site Redeploy Check ===" - echo "GitHub Pages will auto-deploy on the push that follows." - echo "Verifying key files exist and are non-empty:" - ls -l index.html RULES.md projects-inventory.md CHANGELOG.md AUDIT.md - echo "Head of projects-inventory.md (last 10 lines):" - tail -10 projects-inventory.md || true - echo "Redeploy check complete." - - - name: Commit and push updates (with skip ci to prevent loop) - run: | - git add projects-inventory.md CHANGELOG.md AUDIT.md - git commit -m "chore(sync): update projects-inventory.md + CHANGELOG.md + AUDIT.md on merge of #${{ github.event.pull_request.number }} [skip ci]" - git push origin main - - - name: Summary - run: | - echo "::notice title=Rule 2 Extended::Accepted PR #${{ github.event.pull_request.number }} processed: inventory synced, audit logged, site redeploy initiated. All public and append-only." diff --git a/.github/workflows/publish-mcp.yml b/.github/workflows/publish-mcp.yml new file mode 100644 index 0000000..c1e1350 --- /dev/null +++ b/.github/workflows/publish-mcp.yml @@ -0,0 +1,60 @@ +name: Publish to PyPI and MCP Registry + +# Publishes on a version tag, e.g. `git tag v0.1.0 && git push origin v0.1.0`. +# The MCP Registry stores metadata only, so the PyPI package must be published +# first; the registry then verifies ownership via the `mcp-name` marker in the +# package README. +on: + push: + tags: ["v*"] + +jobs: + publish: + runs-on: ubuntu-latest + permissions: + id-token: write # OIDC: PyPI trusted publishing + MCP Registry auth + contents: read + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Sync version from tag into pyproject.toml and server.json + run: | + VERSION="${GITHUB_REF_NAME#v}" + echo "Publishing version $VERSION" + python - "$VERSION" <<'PY' + import json, re, sys + version = sys.argv[1] + # pyproject.toml + pp = open("pyproject.toml").read() + pp = re.sub(r'(?m)^version = ".*"$', f'version = "{version}"', pp, count=1) + open("pyproject.toml", "w").write(pp) + # server.json (top-level + each package version) + data = json.load(open("server.json")) + data["version"] = version + for pkg in data.get("packages", []): + pkg["version"] = version + json.dump(data, open("server.json", "w"), indent=2) + PY + + - name: Build package + run: | + python -m pip install --upgrade build + python -m build + + - name: Publish package to PyPI (trusted publishing / OIDC) + uses: pypa/gh-action-pypi-publish@release/v1 + + - name: Install mcp-publisher + run: | + curl -L "https://github.com/modelcontextprotocol/registry/releases/latest/download/mcp-publisher_$(uname -s | tr '[:upper:]' '[:lower:]')_$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/').tar.gz" | tar xz mcp-publisher + + - name: Authenticate to the MCP Registry (GitHub OIDC) + run: ./mcp-publisher login github-oidc + + - name: Publish metadata to the MCP Registry + run: ./mcp-publisher publish diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..106a5fe --- /dev/null +++ b/.gitignore @@ -0,0 +1,19 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +build/ +dist/ +.venv/ +venv/ +.pytest_cache/ +.ruff_cache/ + +# Secrets and local config +.env +agentsec.yaml +superuser.yaml + +# Runtime artifacts +superuser-audit.log +*.log diff --git a/README.md b/README.md index d083ec7..c52284d 100644 --- a/README.md +++ b/README.md @@ -1,32 +1,191 @@ # Super -**superuser** — A permission-gated superuser (sysadmin) agent exposed over the Model Context Protocol (MCP). + -Gives an AI agent real system-administration powers while wrapping every action in enterprise-grade guardrails: Cisco AI Defense inspection, Cisco Duo Agentic Identity authorization, human approval gates, dangerous-command denylist, append-only audit logging, and Stripe usage-based billing. +A **permission-gated superuser agent** exposed over the [Model Context Protocol +(MCP)](https://modelcontextprotocol.io). It gives an AI agent real +system-administration powers on this machine — running shell commands, checking +system state, and acting on GitHub — while wrapping every action in guardrails: +a dangerous-command denylist, a human-approval gate for risky operations, +least-privilege GitHub access, an append-only audit log, and **Cisco AI Defense** +runtime inspection of every tool call and response. -**One look two fight three listen four break** +The design goal is the version of a "superuser agent" you can actually ship: +powerful, but safe, auditable, and compatible with enterprise guardrails. -The core guard protocol that makes powerful agents shippable and safe. +> **Landing site:** https://fitzyracing1.github.io/super-landing/ (source: [fitzyracing1/super-landing](https://github.com/fitzyracing1/super-landing)). +> The previous Pages site in this repo is preserved at tag `site-v1` / branch `site-archive`. -## Live Site -https://fitzyracing1.github.io/Super/ +## Why it's built this way -## Project +An agent that can do *anything* as root is a liability, not a product. Buyers +(and marketplace security reviews) expect scoping, approvals, and an audit +trail. So capabilities here always pass through a single choke point — the +`Guard` — which enforces, in order: -See the detailed README on the active development branch or the open PRs for full architecture, tools, setup, and verification: +1. **Cisco AI Defense** inspects the tool call (can block). +2. **Cisco Duo Agentic Identity** authorizes the tool call per-policy (can block). +3. **Approval gate** — anything at or above `approval_required_at` (default + `MEDIUM`) needs an explicit `confirm=true` from a human. +4. The operation runs. +5. **Cisco AI Defense** inspects the response before it is returned (catches + data leakage). +6. **Stripe** meters the successful call (best-effort; never blocks). +7. The outcome is written to the **audit log** as + `human owner -> agent -> tool -> action -> outcome`. -- PR #1: Superuser MCP agent with Cisco AI Defense + Duo authz + Stripe billing -- PR #2: Cursor Cloud dev environment + AGENTS.md +This mirrors the enterprise model: your app owns scoping + approvals + local +audit; **Cisco AI Defense** provides runtime threat inspection at the MCP layer; +**Cisco Duo Agentic Identity** provides per-tool-call authorization tied to an +accountable human owner; and **Stripe** turns metered tool calls into revenue. -This GitHub Pages site is the public face of the project. The real implementation lives on feature branches and will be merged to main. +### Cisco Duo Agentic Identity (authorization) -## Quick Links -- [Open PR #1](https://github.com/fitzyracing1/Super/pull/1) -- [Open PR #2](https://github.com/fitzyracing1/Super/pull/2) -- Branch: `cursor/superuser-mcp-agent-1ae7` +Duo follows a "Duo decides, the gateway enforces" model. This agent registers as +a non-human identity, authenticates via the OAuth 2.1 **client-credentials** +grant (with an RFC 8707 resource indicator), and authorizes every tool call +either by matching the tool's required **scope** against what Duo granted, or by +calling an explicit Duo **decision endpoint**. Set `DUO_ENABLED=true` plus +`DUO_TOKEN_URL` / `DUO_CLIENT_ID` / `DUO_CLIENT_SECRET` (see `.env.example`). When +Duo is not configured, calls are allowed but flagged `governed: false`; in +`enforce` mode authorization errors fail closed. -Built for safety at scale. Designed for the frontier. -## Documentation +### Stripe usage-based billing (monetization) -- In-repo docs: [docs/](docs/index.md) -- Wiki: [https://github.com/fitzyracing1/Super/wiki](https://github.com/fitzyracing1/Super/wiki) +Each authorized, successful tool call emits a Stripe **meter event** +(v2 billing meter events API), so the agent can be sold per-action. Configure a +meter in Stripe, then set `STRIPE_ENABLED=true`, `STRIPE_API_KEY`, +`STRIPE_CUSTOMER_ID`, and `STRIPE_METER_EVENT_NAME`. Per-tool weights let +higher-risk actions cost more. Billing is best-effort: metering failures are +logged and never block or fail the tool call (results are annotated +`billed: true|false`). + +## Tools + +| Tool | Risk | Notes | +|------|------|-------| +| `system_status` | LOW | Host, CPU, memory, uptime | +| `disk_usage` | LOW | Per-partition disk usage | +| `list_processes` | LOW | Top processes by memory/cpu | +| `run_command` | HIGH* | Shell command; denylist + approval + audit | +| `github_get_repo` | LOW | Repo metadata (allowlisted repos only) | +| `github_list_issues` | LOW | Issues for an allowlisted repo | +| `github_create_issue_comment` | HIGH | Write; needs approval + `GITHUB_ALLOW_WRITES` | +| `security_status` | — | Reports the active guardrail configuration | + +\* `run_command` is LOW risk only when its base command is in +`command_allowlist`; otherwise HIGH (requires `confirm=true`). + +## Install + +Not yet on PyPI. Install straight from git: + +```bash +pip install "superuser @ git+https://github.com/fitzyracing1/Super.git" +pip install "superuser[all] @ git+https://github.com/fitzyracing1/Super.git" # + Cisco AI Defense + Stripe SDKs +``` + +From a clone: + +Requires Python 3.10+. + +```bash +pip install -e . # core agent (Duo authz uses requests; no extra dep) +pip install -e '.[security]' # + Cisco AI Defense SDK (cisco-aidefense-sdk) +pip install -e '.[billing]' # + Stripe SDK (usage-based billing) +pip install -e '.[all]' # security + billing +pip install -e '.[dev]' # + pytest / ruff +``` + +## Configure + +Copy `.env.example` to `.env` (or use a YAML file — see `superuser.example.yaml`) +and set values. Key settings: + +- `SUPERUSER_HUMAN_OWNER` — the human accountable for the agent's actions. +- `SUPERUSER_APPROVAL_AT` — risk threshold for the approval gate (`LOW`…`CRITICAL`). +- `AI_DEFENSE_*` — Cisco AI Defense credentials/mode (see below). +- `DUO_*` — Cisco Duo Agentic Identity authorization (see below). +- `STRIPE_*` — Stripe usage-based billing (see below). +- `GITHUB_TOKEN`, `GITHUB_ALLOWED_REPOS`, `GITHUB_ALLOW_WRITES` — least-privilege GitHub. + +> For a full console-value → env-var mapping for all three services, see +> [`docs/INTEGRATIONS.md`](docs/INTEGRATIONS.md). + +### Cisco AI Defense + +Install the SDK (`pip install 'superuser[security]'`) and provide credentials: + +- **Explicit inspection** of this server's own tool calls/responses: + set `AI_DEFENSE_INSPECTION_API_KEY`. +- **Runtime protection** (`agentsec.protect()` auto-patches LLM + MCP clients): + set `AI_DEFENSE_API_MODE_LLM_ENDPOINT` + `AI_DEFENSE_API_MODE_LLM_API_KEY`, + or point `AI_DEFENSE_AGENTSEC_CONFIG` at an `agentsec.yaml` + (see `agentsec.example.yaml`). +- `AI_DEFENSE_MODE`: `enforce` (block), `monitor` (log only), or `off`. + +If the SDK is not installed or no credentials are set, the agent still runs but +every result is annotated `protected: false` / `UNPROTECTED` — the gap is never +silent. In `enforce` mode, inspection errors **fail closed** (the call is +blocked); in `monitor` mode they fail open (logged, allowed). + +## Environment variables + +| Area | Variables | +|------|-----------| +| Core | `SUPERUSER_CONFIG`, `SUPERUSER_HUMAN_OWNER`, `SUPERUSER_APPROVAL_AT`, `SUPERUSER_COMMAND_TIMEOUT`, `SUPERUSER_AUDIT_LOG` | +| Cisco AI Defense | `AI_DEFENSE_ENABLED` (default `true`; set `false` for local dev), `AI_DEFENSE_MODE`, `AI_DEFENSE_INSPECTION_API_KEY`, `AI_DEFENSE_API_MODE_LLM_ENDPOINT`, `AI_DEFENSE_API_MODE_LLM_API_KEY`, `AI_DEFENSE_AGENTSEC_CONFIG` | +| Cisco Duo | `DUO_ENABLED`, `DUO_MODE`, `DUO_AGENT_ID`, `DUO_TOKEN_URL`, `DUO_CLIENT_ID`, `DUO_CLIENT_SECRET`, `DUO_AUTHORIZATION_URL`, `DUO_RESOURCE`, `DUO_SCOPES` | +| Stripe | `STRIPE_ENABLED`, `STRIPE_API_KEY`, `STRIPE_CUSTOMER_ID`, `STRIPE_METER_EVENT_NAME`, `STRIPE_DEFAULT_VALUE` | +| GitHub | `GITHUB_TOKEN`, `GITHUB_ALLOWED_REPOS`, `GITHUB_ALLOW_WRITES`, `GITHUB_API_URL` | + +With no keys set the server still runs: Duo and Stripe are off by default, and Cisco AI Defense runs unprotected (clearly flagged) if the SDK/keys are missing. + +## Run + +```bash +# Validate configuration and print guardrail/Cisco status, then exit: +superuser --check + +# Start the MCP server (stdio transport, for MCP clients like Cursor/Claude): +superuser +# or: python -m superuser +``` + +### Register with an MCP client + +Example client config (stdio): + +```json +{ + "mcpServers": { + "superuser": { + "command": "superuser", + "env": { + "SUPERUSER_HUMAN_OWNER": "you@example.com", + "AI_DEFENSE_MODE": "enforce", + "AI_DEFENSE_INSPECTION_API_KEY": "..." + } + } + } +} +``` + +## Test + +```bash +pytest +``` + +## Security notes + +- The dangerous-command denylist is a backstop, not the primary control. The + primary controls are Cisco AI Defense inspection and the approval gate. +- GitHub access **fails closed**: no token or an empty repo allowlist means no + access. Use a fine-grained, least-privilege token. +- Secrets are redacted in the audit log. Keep `.env` / `*.yaml` config files out + of version control (already in `.gitignore`). + +## License + +Apache-2.0. diff --git a/agentsec.example.yaml b/agentsec.example.yaml new file mode 100644 index 0000000..50b47ee --- /dev/null +++ b/agentsec.example.yaml @@ -0,0 +1,16 @@ +# Cisco AI Defense Agent Runtime (agentsec) configuration. +# Copy to agentsec.yaml and point AI_DEFENSE_AGENTSEC_CONFIG at it, or let the +# agent build API-mode config from environment variables instead. +# +# Values support ${VAR} substitution from the environment. + +api_mode: + llm: + # "enforce" blocks policy violations; "monitor" logs only; "off" disables. + mode: enforce + endpoint: ${AI_DEFENSE_API_MODE_LLM_ENDPOINT} + api_key: ${AI_DEFENSE_API_MODE_LLM_API_KEY} + mcp: + mode: enforce + endpoint: ${AI_DEFENSE_API_MODE_MCP_ENDPOINT} + api_key: ${AI_DEFENSE_API_MODE_MCP_API_KEY} diff --git a/docs/INTEGRATIONS.md b/docs/INTEGRATIONS.md new file mode 100644 index 0000000..3d65324 --- /dev/null +++ b/docs/INTEGRATIONS.md @@ -0,0 +1,122 @@ +# Integration setup: Cisco AI Defense, Cisco Duo, Stripe + +This document maps each external service's console values to the agent's +environment variables. **Do not commit real secrets or tenant-specific IDs** — +put them in a `.env` file (gitignored) locally, or in Cursor **Cloud Agents → +Secrets** for cloud runs. Placeholders below use ``. + +Verify any setup with: + +```bash +superuser --check +``` + +--- + +## Cisco AI Defense (runtime threat inspection) + +Console: **Cisco Security Cloud** (`security.cisco.com`; API at +`api.security.cisco.com`). Try-before-you-buy via **AI Defense: Explorer Edition**. + +Install the SDK: `pip install 'superuser[security]'`. + +| Console value | Env var | +|---|---| +| Inspection API key | `AI_DEFENSE_INSPECTION_API_KEY` | +| Agent Runtime (API mode) endpoint | `AI_DEFENSE_API_MODE_LLM_ENDPOINT` | +| Agent Runtime (API mode) key | `AI_DEFENSE_API_MODE_LLM_API_KEY` | +| — (enable/disable) | `AI_DEFENSE_ENABLED=true` | +| — (`enforce`\|`monitor`\|`off`) | `AI_DEFENSE_MODE=enforce` | +| Path to an `agentsec.yaml` (alt to endpoint/key) | `AI_DEFENSE_AGENTSEC_CONFIG` | + +`monitor` logs violations but allows calls; `enforce` blocks them and fails +closed on inspection errors. + +--- + +## Cisco Duo Agentic Identity (per-tool-call authorization) + +Console: **Duo Admin Panel** (`admin.duosecurity.com`). No extra Python +dependency — the agent uses OAuth 2.1 client-credentials over `requests`. + +### Where to find the values + +1. **Configure Single Sign-On (SSO)** must be set up first (the OAuth 2.1 + authorization server lives inside Duo SSO). +2. **Application Catalog → search "OAuth 2.1 / OIDC"** → create the app. +3. **Metadata** tab → **Token URL** and **Issuer**. +4. **Scopes** tab → create `superuser.read`, `superuser.exec`, `superuser.write`. +5. **Clients** tab → **Client 1** (a confidential client) → + **Client ID**, and **Reset Client Secret** to reveal the secret **once** + (do this on a **desktop** browser; the reveal often fails on mobile). + +| Console value | Env var | +|---|---| +| — (enable/disable) | `DUO_ENABLED=true` | +| — (`enforce`\|`monitor`\|`off`) | `DUO_MODE=enforce` | +| A name/id for this agent identity | `DUO_AGENT_ID=` | +| Metadata → Token URL | `DUO_TOKEN_URL=https://sso-.sso.duosecurity.com/oauth2//token` | +| Clients → Client 1 → Client ID | `DUO_CLIENT_ID=` | +| Clients → Client 1 → Client Secret (reset to reveal) | `DUO_CLIENT_SECRET=` | +| Scopes (space-separated) | `DUO_SCOPES=superuser.read superuser.exec superuser.write` | +| RFC 8707 resource indicator (audience), optional | `DUO_RESOURCE=` | +| Explicit decision endpoint, optional | `DUO_AUTHORIZATION_URL=` | + +### Scope → tool mapping + +Set under `duo.tool_scopes` in `superuser.yaml` (see `superuser.example.yaml`): + +| Tool | Required scope | +|---|---| +| `system_status`, `disk_usage`, `list_processes` | `superuser.read` | +| `github_get_repo`, `github_list_issues` | `superuser.read` | +| `run_command` | `superuser.exec` | +| `github_create_issue_comment` | `superuser.write` | + +### Notes + +- The client must support the **Client Credentials** grant (machine-to-machine). +- The **Client Secret is shown only once**, at reset time — copy it immediately. +- Without Duo configured, calls run but are flagged `governed: false`. +- `monitor` logs would-be denials without blocking; `enforce` blocks and fails + closed on authorization errors. + +--- + +## Stripe (usage-based billing) + +Console: **Stripe Dashboard** (`dashboard.stripe.com`). Create a **billing +meter** whose event name matches `STRIPE_METER_EVENT_NAME`. + +Install the SDK: `pip install 'superuser[billing]'`. + +| Console value | Env var | +|---|---| +| — (enable/disable) | `STRIPE_ENABLED=true` | +| Secret API key | `STRIPE_API_KEY=sk_live_<...>` | +| Customer id to bill | `STRIPE_CUSTOMER_ID=cus_<...>` | +| Meter event name | `STRIPE_METER_EVENT_NAME=superuser_tool_call` | +| Default units per call | `STRIPE_DEFAULT_VALUE=1` | + +Per-tool billing weights go under `stripe.per_tool_values` in `superuser.yaml` +(e.g. charge more for `run_command`). Billing is best-effort: metering failures +are logged and never block or fail a tool call (results annotated +`billed: true|false`). + +--- + +## Quick local test without any credentials + +Everything degrades gracefully, so you can exercise the full guard pipeline with +no external services: + +```bash +DUO_ENABLED=true DUO_MODE=monitor \ +AI_DEFENSE_ENABLED=false \ +SUPERUSER_HUMAN_OWNER="you@example.com" \ +superuser --check +``` + +Then register the server with an MCP client (see `README.md`) and call tools; +results will show `protected/governed/billed: false` until the respective +services are configured. diff --git a/index.html b/index.html index 7b7c22b..405e810 100644 --- a/index.html +++ b/index.html @@ -1,459 +1,6 @@ - - - - - - - - superuser • One look two fight three listen four break - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
-
- -
- -
-
- - GUARD PROTOCOL -
-

- One look
two fight
three listen
four break -

-

The four-phase security mantra that turns a dangerous superuser into a governable, auditable, enterprise-ready agent.

-
- -
-
-
- - MCP • Model Context Protocol -
-
Python 3.10+
-
- -

- superuser
MCP agent -

- -

Real sysadmin powers for AI agents.
Wrapped in layers that actually ship.

- -
- - See the Guard Pipeline - - - - - View on GitHub - -
- -
-
-
- 16 tests passing • ruff clean -
- - -
-
-
-
- - -
-
-
THE CORE MANTRA
-

One look two fight three listen four break

-

This is not marketing copy. It is the literal sequence every tool call must survive before it reaches your machine or GitHub.

-
- -
-
-
01
-
LOOK
-
Cisco AI Defense inspects the incoming tool call for threats, prompt injection, or policy violations.
-
-
-
02
-
FIGHT
-
Duo Agentic Identity + policy engine authorizes (or blocks) based on scope, human owner, and risk level.
-
-
-
03
-
LISTEN
-
Approval gate + append-only audit log. Risky actions require explicit human confirm=true.
-
-
-
04
-
BREAK
-
Dangerous commands are denied by denylist. Leaking responses are caught by post-execution AI Defense inspection.
-
-
-
- - -
-
-
THE CHOKE POINT
-

Every action passes through one Guard.

-

No tool runs unless it survives the full layered pipeline. The agent is powerful exactly because it is heavily constrained.

-
- -
-
-
1
-
-
Cisco AI Defense
-
Runtime inspection of tool calls + responses. Can block in enforce mode. Degrades gracefully with annotation.
-
-
-
-
2
-
-
Cisco Duo Agentic Identity
-
Per-tool-call authorization tied to a real human owner via OAuth 2.1 client-credentials. Fails closed.
-
-
-
-
3
-
-
Approval Gate + Denylist
-
Anything ≥ MEDIUM risk requires explicit human confirmation. Hard denylist for obviously destructive commands.
-
-
-
-
4
-
-
Stripe Billing + Audit Log
-
Usage metered per successful call (best-effort). Every decision recorded as immutable human → agent → tool → outcome.
-
-
-
-
- - -
-
-
EXPOSED TOOLS
-

Powerful capabilities.
Strictly gated.

-
- -
-
-
system_status / disk_usage / list_processes
-
Read-only system inspection. Always LOW risk.
-
LOW
-
-
-
run_command
-
Full shell execution with denylist, approval gate, and post-run inspection. HIGH unless allowlisted.
-
HIGH (with guardrails)
-
-
-
github_* tools
-
get_repo, list_issues, create_issue_comment. Least-privilege + allowlist only. Write actions require approval.
-
LOW / HIGH
-
-
-
security_status
-
Live report of all active guardrails, AI Defense mode, Duo governance, billing status, and config health.
-
—
-
-
-
- - -
-
-
-
GET STARTED IN SECONDS
-

Install • Configure • Run

-
- -
-
-
# Install
- 
pip install -e '.[all]'
-# or with dev tools
-pip install -e '.[all,dev]'
-
Requires Python 3.10+. Optional: cisco-aidefense-sdk + stripe
-
-
-
# Run the MCP server
- 
superuser --check   # validate config
-superuser           # start stdio MCP server
-
16 tests • ruff clean • end-to-end verified
-
-
- -
See .env.example, superuser.example.yaml, and open PRs for full integration setup (Cisco, Duo, Stripe).
-
-
- - -
-
-
- - SUPPORT & COMMERCIAL -
-

Contact Sales & Help

-

Questions about licensing, enterprise deployments, custom guard configurations, or just need help getting started with the protocol?

-
- -
- -
-
-
- -
-

Need Help?

-
-

The superuser project is fully documented with a formal ruleset and contribution process. All changes go through the guard protocol for safety and auditability.

- -
- - - Read RULES.md - - - - Contributing - -
- - Open or browse GitHub Issues - -
- - -
-
PRICING & ENTERPRISE
- -
-
- -
-

Contact Sales

-
- -

For pricing, enterprise licensing, priority support, custom integrations, or volume deployments of the superuser MCP agent.

-

Direct line to the maintainer. Fast responses for commercial inquiries.

- - - Email for Pricing - - - -
- - fitzyracing1@gmail.com -
-
-
- -
- All commercial discussions respect the open-source core and the four-phase guard protocol. Public transparency is non-negotiable. -
-
- - -
-
-
- - PUBLIC DEPLOY ON MERGE -
- -

On accepted PR → we write, it stays public.

- -

Edit the site or agent code, open a PR, get it accepted and merged to main. GitHub Pages automatically redeploys the updated site within minutes. Everything remains fully public and auditable.

- -
- - - View the full listed Ruleset (RULES.md) - -
- -
- - Read CONTRIBUTING.md - - - View Open PRs - -
-
-
- - - - - - - \ No newline at end of file + + +Super — moved + + +

The Super landing site has moved to fitzyracing1.github.io/super-landing. Code: github.com/fitzyracing1/Super.

diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..09569ce --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,39 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "superuser" +version = "0.1.0" +description = "A permission-gated superuser (sysadmin) MCP agent with Cisco AI Defense runtime guardrails." +readme = "README.md" +requires-python = ">=3.10" +license = { text = "Apache-2.0" } +authors = [{ name = "Super" }] +keywords = ["mcp", "agent", "sysadmin", "superuser", "cisco", "ai-defense", "guardrails"] + +dependencies = [ + "mcp>=1.2.0,<2", + "psutil>=5.9", + "PyYAML>=6.0", + "requests>=2.31", +] + +[project.optional-dependencies] +# Cisco AI Defense runtime guardrails. Imported lazily so the server still +# runs (in an unprotected, clearly-warned state) when this is not installed. +security = ["cisco-aidefense-sdk>=2.1.0"] +# Stripe usage-based billing. Optional; metering degrades gracefully without it. +billing = ["stripe>=11.0"] +dev = ["pytest>=8.0", "ruff>=0.6"] +all = ["cisco-aidefense-sdk>=2.1.0", "stripe>=11.0"] + +[project.scripts] +superuser = "superuser.__main__:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.ruff] +line-length = 100 +target-version = "py310" diff --git a/server.json b/server.json new file mode 100644 index 0000000..0c99d56 --- /dev/null +++ b/server.json @@ -0,0 +1,71 @@ +{ + "$schema": "https://static.modelcontextprotocol.io/schemas/2025-12-11/server.schema.json", + "name": "io.github.fitzyracing1/superuser", + "description": "Permission-gated superuser (sysadmin) MCP agent with enterprise guardrails and audit.", + "repository": { + "url": "https://github.com/fitzyracing1/Super", + "source": "github" + }, + "version": "0.1.0", + "packages": [ + { + "registryType": "pypi", + "identifier": "superuser", + "version": "0.1.0", + "transport": { + "type": "stdio" + }, + "environmentVariables": [ + { + "name": "SUPERUSER_HUMAN_OWNER", + "description": "Human accountable for the agent's actions (recorded in every audit entry).", + "isRequired": false, + "isSecret": false, + "format": "string" + }, + { + "name": "SUPERUSER_APPROVAL_AT", + "description": "Minimum risk level that triggers the human-approval gate: LOW, MEDIUM, HIGH, or CRITICAL.", + "isRequired": false, + "isSecret": false, + "format": "string" + }, + { + "name": "AI_DEFENSE_INSPECTION_API_KEY", + "description": "Cisco AI Defense Inspection API key for runtime tool-call/response inspection.", + "isRequired": false, + "isSecret": true, + "format": "string" + }, + { + "name": "DUO_TOKEN_URL", + "description": "Cisco Duo OAuth 2.1 token endpoint (client-credentials) for per-tool-call authorization.", + "isRequired": false, + "isSecret": false, + "format": "string" + }, + { + "name": "DUO_CLIENT_ID", + "description": "Cisco Duo OAuth client ID for the agent's non-human identity.", + "isRequired": false, + "isSecret": false, + "format": "string" + }, + { + "name": "DUO_CLIENT_SECRET", + "description": "Cisco Duo OAuth client secret.", + "isRequired": false, + "isSecret": true, + "format": "string" + }, + { + "name": "STRIPE_API_KEY", + "description": "Stripe secret API key for usage-based (metered) billing.", + "isRequired": false, + "isSecret": true, + "format": "string" + } + ] + } + ] +} diff --git a/src/superuser/__init__.py b/src/superuser/__init__.py new file mode 100644 index 0000000..4e309f1 --- /dev/null +++ b/src/superuser/__init__.py @@ -0,0 +1,5 @@ +"""Superuser: a permission-gated sysadmin MCP agent with Cisco AI Defense guardrails.""" + +__version__ = "0.1.0" + +__all__ = ["__version__"] diff --git a/src/superuser/__main__.py b/src/superuser/__main__.py new file mode 100644 index 0000000..cb7b210 --- /dev/null +++ b/src/superuser/__main__.py @@ -0,0 +1,62 @@ +"""Entry point: `superuser` / `python -m superuser`.""" + +from __future__ import annotations + +import argparse +import logging +import sys + +from .config import Config +from .server import build_server + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="superuser", + description="Permission-gated superuser MCP agent with Cisco AI Defense guardrails.", + ) + parser.add_argument("--config", help="Path to a YAML config file (or set SUPERUSER_CONFIG).") + parser.add_argument( + "--transport", default="stdio", choices=["stdio", "sse", "streamable-http"], + help="MCP transport (default: stdio).", + ) + parser.add_argument("--log-level", default="INFO") + parser.add_argument( + "--check", action="store_true", + help="Validate config and print security status, then exit (no server).", + ) + args = parser.parse_args(argv) + + logging.basicConfig( + level=getattr(logging, args.log_level.upper(), logging.INFO), + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + stream=sys.stderr, + ) + + config = Config.load(args.config) + server = build_server(config) + + if args.check: + from .authz import DuoAuthorizer + from .billing import StripeMeter + from .security import CiscoAIDefense + + defense = CiscoAIDefense(config.cisco) + authorizer = DuoAuthorizer(config.duo) + meter = StripeMeter(config.stripe) + print("Superuser agent configuration OK.") + print(f" Human owner: {config.human_owner}") + print(f" Approval required at: {config.approval_required_at.name}") + print(f" Audit log: {config.audit_log_path}") + print(f" Cisco AI Defense: {defense.status}") + print(f" Cisco Duo (authz): {authorizer.status}") + print(f" Stripe billing: {meter.status}") + print(f" GitHub configured: {bool(config.github.token)}") + return 0 + + server.run(transport=args.transport) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/superuser/audit.py b/src/superuser/audit.py new file mode 100644 index 0000000..a957aa9 --- /dev/null +++ b/src/superuser/audit.py @@ -0,0 +1,70 @@ +"""Append-only audit logging. + +Every tool invocation produces an identity-correlated record: +``human owner -> agent -> tool -> action -> outcome``. This is the local +counterpart to the audit chain enforced by Cisco Duo's MCP gateway, and it is +what makes the agent's actions reviewable after the fact. +""" + +from __future__ import annotations + +import json +import os +import threading +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +# Argument keys whose values must never be written to the audit log verbatim. +_SENSITIVE_KEYS = {"token", "api_key", "apikey", "password", "secret", "authorization"} + + +def _redact(value: Any) -> Any: + if isinstance(value, dict): + return { + k: ("***REDACTED***" if k.lower() in _SENSITIVE_KEYS else _redact(v)) + for k, v in value.items() + } + if isinstance(value, list): + return [_redact(v) for v in value] + if isinstance(value, str) and len(value) > 2000: + return value[:2000] + f"...[truncated {len(value) - 2000} chars]" + return value + + +class AuditLog: + """Thread-safe JSONL audit logger.""" + + def __init__(self, path: str | os.PathLike[str], human_owner: str = "unknown") -> None: + self.path = Path(path) + self.human_owner = human_owner + self._lock = threading.Lock() + if self.path.parent and not self.path.parent.exists(): + self.path.parent.mkdir(parents=True, exist_ok=True) + + def record( + self, + tool: str, + arguments: dict[str, Any], + decision: str, + status: str, + *, + risk: str | None = None, + detail: str | None = None, + ) -> dict[str, Any]: + entry = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "human_owner": self.human_owner, + "agent": "superuser", + "tool": tool, + "arguments": _redact(arguments), + "risk": risk, + "decision": decision, + "status": status, + "detail": detail, + } + line = json.dumps(entry, ensure_ascii=False) + with self._lock: + with self.path.open("a", encoding="utf-8") as fh: + fh.write(line + "\n") + return entry diff --git a/src/superuser/authz.py b/src/superuser/authz.py new file mode 100644 index 0000000..44e5a32 --- /dev/null +++ b/src/superuser/authz.py @@ -0,0 +1,155 @@ +"""Cisco Duo Agentic Identity authorization. + +Implements the client side of Duo's "Duo decides, the gateway enforces" model +for per-tool-call authorization: + +- The agent authenticates as a registered non-human identity using the OAuth 2.1 + client-credentials grant, requesting scopes (and, per RFC 8707, a resource + indicator naming the audience it intends to access). +- Before each tool runs, the call is authorized either by an explicit Duo + decision endpoint (``authorization_url``) or, by default, by checking the + tool's required scope against the scopes Duo granted in the token. + +Only ``requests`` is needed (no extra SDK). When Duo is not configured the +authorizer degrades gracefully: calls are allowed but flagged UNGOVERNED. In +``enforce`` mode, authorization errors fail closed; in ``monitor`` they are +logged and allowed. +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass + +import requests + +from .config import DuoConfig + +logger = logging.getLogger("superuser.authz") + + +@dataclass +class AuthzDecision: + allowed: bool + governed: bool + reason: str + + @classmethod + def ungoverned(cls) -> "AuthzDecision": + return cls(allowed=True, governed=False, reason="Duo not configured") + + @classmethod + def allow(cls, reason: str = "authorized") -> "AuthzDecision": + return cls(allowed=True, governed=True, reason=reason) + + @classmethod + def deny(cls, reason: str) -> "AuthzDecision": + return cls(allowed=False, governed=True, reason=reason) + + +class DuoAuthorizer: + def __init__(self, config: DuoConfig) -> None: + self.config = config + self._token: str | None = None + self._granted_scopes: set[str] = set() + self._token_expiry: float = 0.0 + + def _fetch_token(self) -> str | None: + """Client-credentials grant; caches the token until shortly before expiry.""" + now = time.time() + if self._token and now < self._token_expiry: + return self._token + + data = {"grant_type": "client_credentials"} + if self.config.scopes: + data["scope"] = " ".join(self.config.scopes) + if self.config.resource: + data["resource"] = self.config.resource # RFC 8707 + + resp = requests.post( + self.config.token_url, + data=data, + auth=(self.config.client_id, self.config.client_secret), + headers={"Accept": "application/json"}, + timeout=15, + ) + if resp.status_code >= 400: + raise RuntimeError(f"Duo token endpoint {resp.status_code}: {resp.text[:300]}") + + body = resp.json() + self._token = body.get("access_token") + granted = body.get("scope", "") + self._granted_scopes = set(granted.split()) if granted else set(self.config.scopes) + self._token_expiry = now + int(body.get("expires_in", 300)) - 30 + return self._token + + def authorize(self, tool_name: str, arguments: dict, risk: str) -> AuthzDecision: + if not self.config.can_enforce: + return AuthzDecision.ungoverned() + try: + token = self._fetch_token() + if not token: + raise RuntimeError("no access token returned by Duo") + + if self.config.authorization_url: + return self._authorize_via_endpoint(token, tool_name, arguments, risk) + return self._authorize_via_scopes(tool_name) + except Exception as exc: + logger.error("Duo authorization error for %s: %s", tool_name, exc) + if self.config.mode == "monitor": + return AuthzDecision(True, True, f"authorization error (monitor): {exc}") + return AuthzDecision.deny(f"authorization error: {exc}") + + def _authorize_via_scopes(self, tool_name: str) -> AuthzDecision: + required = self.config.tool_scopes.get(tool_name) + if required is None: + # No scope mapping for this tool: allow but note it's unmapped. + return AuthzDecision.allow("no scope requirement mapped") + if required in self._granted_scopes: + return AuthzDecision.allow(f"scope '{required}' granted") + reason = f"missing required scope '{required}'" + if self.config.mode == "monitor": + logger.warning("Duo (monitor) would deny %s: %s", tool_name, reason) + return AuthzDecision(True, True, f"{reason} (monitor)") + return AuthzDecision.deny(reason) + + def _authorize_via_endpoint( + self, token: str, tool_name: str, arguments: dict, risk: str + ) -> AuthzDecision: + resp = requests.post( + self.config.authorization_url, + headers={"Authorization": f"Bearer {token}", "Accept": "application/json"}, + json={ + "agent_id": self.config.agent_id, + "resource": self.config.resource, + "tool": tool_name, + "action": tool_name, + "risk": risk, + "arguments": arguments, + }, + timeout=15, + ) + if resp.status_code == 403: + return AuthzDecision.deny("Duo policy denied this tool call") + if resp.status_code >= 400: + raise RuntimeError(f"Duo decision endpoint {resp.status_code}: {resp.text[:300]}") + + body = resp.json() if resp.text else {} + decision = str(body.get("decision", body.get("action", "allow"))).lower() + if decision in ("allow", "permit", "allowed"): + return AuthzDecision.allow(body.get("reason", "authorized by Duo")) + reason = body.get("reason", "denied by Duo policy") + if self.config.mode == "monitor": + return AuthzDecision(True, True, f"{reason} (monitor)") + return AuthzDecision.deny(reason) + + @property + def status(self) -> dict: + return { + "enabled": self.config.enabled, + "mode": self.config.mode, + "enforcing": self.config.can_enforce, + "agent_id": self.config.agent_id, + "authorization": "endpoint" if self.config.authorization_url else "scopes", + } diff --git a/src/superuser/billing.py b/src/superuser/billing.py new file mode 100644 index 0000000..3c8a401 --- /dev/null +++ b/src/superuser/billing.py @@ -0,0 +1,91 @@ +"""Stripe usage-based (metered) billing. + +After a tool call is authorized and executed successfully, the agent reports a +Stripe meter event so it can be sold per-action. This uses Stripe's v2 billing +meter events API via the official ``stripe`` SDK. + +Billing is strictly best-effort: it must never block or fail a tool call. If the +SDK is missing, no key is configured, or Stripe returns an error, the failure is +logged and the tool result is returned unchanged (annotated ``billed: false``). +""" + +from __future__ import annotations + +import logging +import uuid +from dataclasses import dataclass + +from .config import StripeConfig + +logger = logging.getLogger("superuser.billing") + + +@dataclass +class BillingResult: + billed: bool + value: int = 0 + identifier: str | None = None + detail: str | None = None + + +class StripeMeter: + def __init__(self, config: StripeConfig) -> None: + self.config = config + self._client = None + self._available = self._probe_sdk() + if config.enabled and not self._available: + logger.warning( + "Stripe SDK not installed; usage will not be metered. " + "Install with: pip install 'superuser[billing]'" + ) + + @staticmethod + def _probe_sdk() -> bool: + try: + import stripe # noqa: F401 + except Exception: + return False + return True + + def _get_client(self): + if self._client is not None: + return self._client + if not (self._available and self.config.can_bill): + return None + import stripe + + self._client = stripe.StripeClient(self.config.api_key) + return self._client + + def record(self, tool_name: str) -> BillingResult: + if not self.config.enabled: + return BillingResult(billed=False, detail="billing disabled") + client = self._get_client() + if client is None: + return BillingResult(billed=False, detail="billing not configured") + + value = self.config.value_for(tool_name) + identifier = f"{tool_name}:{uuid.uuid4()}" + try: + client.v2.billing.meter_events.create({ + "event_name": self.config.event_name, + "identifier": identifier, + "payload": { + "stripe_customer_id": self.config.customer_id, + "value": str(value), + "tool": tool_name, + }, + }) + return BillingResult(billed=True, value=value, identifier=identifier) + except Exception as exc: # billing must never break the tool call + logger.error("Stripe meter event failed for %s: %s", tool_name, exc) + return BillingResult(billed=False, value=value, detail=str(exc)) + + @property + def status(self) -> dict: + return { + "sdk_installed": self._available, + "enabled": self.config.enabled, + "configured": bool(self.config.can_bill), + "event_name": self.config.event_name, + } diff --git a/src/superuser/config.py b/src/superuser/config.py new file mode 100644 index 0000000..935c669 --- /dev/null +++ b/src/superuser/config.py @@ -0,0 +1,314 @@ +"""Configuration for the superuser agent. + +Settings are resolved with the following precedence (highest first): +1. Explicit values in a YAML config file (path from ``SUPERUSER_CONFIG``). +2. Environment variables. +3. Built-in safe defaults. + +The defaults are deliberately conservative: manual approval is required for +anything above low risk, and GitHub / shell writes are locked down until the +operator opts in. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from enum import IntEnum +from pathlib import Path +from typing import Any + +import yaml + + +class Risk(IntEnum): + """Relative risk of a tool invocation. Higher means more dangerous.""" + + LOW = 0 + MEDIUM = 1 + HIGH = 2 + CRITICAL = 3 + + @classmethod + def parse(cls, value: Any) -> "Risk": + if isinstance(value, Risk): + return value + if isinstance(value, int): + return cls(value) + return cls[str(value).strip().upper()] + + +# Commands that are never allowed to run, regardless of approval. These are +# blunt substring/word checks — a defense-in-depth backstop, not the primary +# control (Cisco AI Defense + approval gating are the primary controls). +DEFAULT_COMMAND_DENYLIST: tuple[str, ...] = ( + "rm -rf /", + "mkfs", + ":(){:|:&};:", # fork bomb + "dd if=/dev/zero", + "dd if=/dev/random", + "> /dev/sda", + "chmod -r 000 /", + "shutdown", + "reboot", + "halt", + "poweroff", +) + + +def _env(name: str, default: str | None = None) -> str | None: + value = os.environ.get(name) + return value if value not in (None, "") else default + + +def _env_bool(name: str, default: bool) -> bool: + raw = _env(name) + if raw is None: + return default + return raw.strip().lower() in ("1", "true", "yes", "on") + + +@dataclass +class CiscoConfig: + """Cisco AI Defense runtime-guardrail settings. + + ``mode`` mirrors the AI Defense Agent Runtime SDK: + - ``enforce``: block requests/responses that violate policy. + - ``monitor``: log violations but allow the call through. + - ``off``: do not call AI Defense at all. + """ + + enabled: bool = True + mode: str = "enforce" + # Inspection-API key used for explicit MCP tool-call inspection. + inspection_api_key: str | None = None + # agentsec.protect() API-mode endpoint/key for auto-patched LLM+MCP calls. + runtime_endpoint: str | None = None + runtime_api_key: str | None = None + # Path to an agentsec.yaml; if present, it takes precedence for protect(). + agentsec_config_path: str | None = None + + @property + def can_inspect(self) -> bool: + return self.enabled and self.mode != "off" and bool(self.inspection_api_key) + + @property + def can_protect(self) -> bool: + return self.enabled and self.mode != "off" and bool( + self.agentsec_config_path or (self.runtime_endpoint and self.runtime_api_key) + ) + + +@dataclass +class DuoConfig: + """Cisco Duo Agentic Identity authorization settings. + + Duo follows a "Duo decides, the gateway enforces" model: per-tool-call + authorization is evaluated against Duo's policy engine, keyed to the agent's + registered identity and its accountable human owner. This agent + authenticates via the OAuth 2.1 client-credentials grant (machine-to-machine) + and checks each tool call against the scopes/policy Duo grants. + + ``mode`` mirrors the other guardrails: ``enforce`` blocks unauthorized + calls, ``monitor`` logs them but allows, ``off`` disables. + """ + + enabled: bool = False + mode: str = "enforce" + agent_id: str | None = None + token_url: str | None = None + client_id: str | None = None + client_secret: str | None = None + # Optional explicit per-tool-call decision endpoint. When unset, the granted + # OAuth scopes are used to authorize (scope-based access control). + authorization_url: str | None = None + # RFC 8707 resource indicator (the audience this agent intends to access). + resource: str | None = None + # Scopes requested at token time and/or matched per tool. + scopes: list[str] = field(default_factory=list) + # Map of tool name -> scope required to invoke it. + tool_scopes: dict[str, str] = field(default_factory=dict) + + @property + def can_enforce(self) -> bool: + return ( + self.enabled + and self.mode != "off" + and bool(self.token_url and self.client_id and self.client_secret) + ) + + +@dataclass +class StripeConfig: + """Stripe usage-based (metered) billing settings. + + A successful, authorized tool call emits a Stripe meter event so the agent + can be sold on a per-action basis. Billing never blocks or fails a tool + call — metering errors are logged and the action still returns. + """ + + enabled: bool = False + api_key: str | None = None + customer_id: str | None = None + event_name: str = "superuser_tool_call" + default_value: int = 1 + # Optional per-tool billing weights (e.g. high-risk actions cost more). + per_tool_values: dict[str, int] = field(default_factory=dict) + + @property + def can_bill(self) -> bool: + return self.enabled and bool(self.api_key and self.customer_id) + + def value_for(self, tool_name: str) -> int: + return int(self.per_tool_values.get(tool_name, self.default_value)) + + +@dataclass +class GitHubConfig: + """Least-privilege GitHub access settings.""" + + token: str | None = None + api_url: str = "https://api.github.com" + # Empty means no repositories are permitted (fail closed). + allowed_repos: list[str] = field(default_factory=list) + allow_writes: bool = False + + def repo_allowed(self, repo: str) -> bool: + if not self.allowed_repos: + return False + if "*" in self.allowed_repos: + return True + return repo in self.allowed_repos + + +@dataclass +class Config: + """Top-level agent configuration.""" + + approval_required_at: Risk = Risk.MEDIUM + command_allowlist: list[str] = field(default_factory=list) + command_denylist: list[str] = field(default_factory=lambda: list(DEFAULT_COMMAND_DENYLIST)) + command_timeout_seconds: int = 60 + audit_log_path: Path = field(default_factory=lambda: Path("superuser-audit.log")) + human_owner: str = "unknown" + cisco: CiscoConfig = field(default_factory=CiscoConfig) + duo: DuoConfig = field(default_factory=DuoConfig) + stripe: StripeConfig = field(default_factory=StripeConfig) + github: GitHubConfig = field(default_factory=GitHubConfig) + + def requires_approval(self, risk: Risk) -> bool: + return risk >= self.approval_required_at + + @classmethod + def load(cls, config_path: str | os.PathLike[str] | None = None) -> "Config": + cfg = cls.from_env() + path = config_path or _env("SUPERUSER_CONFIG") + if path and Path(path).is_file(): + cfg._apply_yaml(yaml.safe_load(Path(path).read_text()) or {}) + return cfg + + @classmethod + def from_env(cls) -> "Config": + cfg = cls() + cfg.human_owner = _env("SUPERUSER_HUMAN_OWNER", cfg.human_owner) + approval = _env("SUPERUSER_APPROVAL_AT") + if approval: + cfg.approval_required_at = Risk.parse(approval) + cfg.command_timeout_seconds = int( + _env("SUPERUSER_COMMAND_TIMEOUT", str(cfg.command_timeout_seconds)) + ) + audit = _env("SUPERUSER_AUDIT_LOG") + if audit: + cfg.audit_log_path = Path(audit) + + cfg.cisco = CiscoConfig( + enabled=_env_bool("AI_DEFENSE_ENABLED", True), + mode=_env("AI_DEFENSE_MODE", "enforce"), + inspection_api_key=_env("AI_DEFENSE_INSPECTION_API_KEY"), + runtime_endpoint=_env("AI_DEFENSE_API_MODE_LLM_ENDPOINT"), + runtime_api_key=_env("AI_DEFENSE_API_MODE_LLM_API_KEY"), + agentsec_config_path=_env("AI_DEFENSE_AGENTSEC_CONFIG"), + ) + + duo_scopes = _env("DUO_SCOPES", "") + cfg.duo = DuoConfig( + enabled=_env_bool("DUO_ENABLED", False), + mode=_env("DUO_MODE", "enforce"), + agent_id=_env("DUO_AGENT_ID"), + token_url=_env("DUO_TOKEN_URL"), + client_id=_env("DUO_CLIENT_ID"), + client_secret=_env("DUO_CLIENT_SECRET"), + authorization_url=_env("DUO_AUTHORIZATION_URL"), + resource=_env("DUO_RESOURCE"), + scopes=[s.strip() for s in (duo_scopes or "").split() if s.strip()], + ) + + cfg.stripe = StripeConfig( + enabled=_env_bool("STRIPE_ENABLED", False), + api_key=_env("STRIPE_API_KEY"), + customer_id=_env("STRIPE_CUSTOMER_ID"), + event_name=_env("STRIPE_METER_EVENT_NAME", "superuser_tool_call"), + default_value=int(_env("STRIPE_DEFAULT_VALUE", "1")), + ) + + repos = _env("GITHUB_ALLOWED_REPOS", "") + cfg.github = GitHubConfig( + token=_env("GITHUB_TOKEN"), + api_url=_env("GITHUB_API_URL", "https://api.github.com"), + allowed_repos=[r.strip() for r in (repos or "").split(",") if r.strip()], + allow_writes=_env_bool("GITHUB_ALLOW_WRITES", False), + ) + return cfg + + def _apply_yaml(self, data: dict[str, Any]) -> None: + if "approval_required_at" in data: + self.approval_required_at = Risk.parse(data["approval_required_at"]) + if "command_allowlist" in data: + self.command_allowlist = list(data["command_allowlist"] or []) + if "command_denylist" in data: + # Extend rather than replace, so operators can't accidentally drop + # the built-in dangerous-command backstops. + self.command_denylist = list(DEFAULT_COMMAND_DENYLIST) + list( + data["command_denylist"] or [] + ) + if "command_timeout_seconds" in data: + self.command_timeout_seconds = int(data["command_timeout_seconds"]) + if "audit_log_path" in data: + self.audit_log_path = Path(data["audit_log_path"]) + if "human_owner" in data: + self.human_owner = str(data["human_owner"]) + + cisco = data.get("cisco") or {} + for key in ("enabled", "mode", "inspection_api_key", "runtime_endpoint", + "runtime_api_key", "agentsec_config_path"): + if key in cisco: + setattr(self.cisco, key, cisco[key]) + + duo = data.get("duo") or {} + for key in ("enabled", "mode", "agent_id", "token_url", "client_id", + "client_secret", "authorization_url", "resource"): + if key in duo: + setattr(self.duo, key, duo[key]) + if "scopes" in duo: + self.duo.scopes = list(duo["scopes"] or []) + if "tool_scopes" in duo: + self.duo.tool_scopes = dict(duo["tool_scopes"] or {}) + + stripe = data.get("stripe") or {} + for key in ("enabled", "api_key", "customer_id", "event_name", "default_value"): + if key in stripe: + setattr(self.stripe, key, stripe[key]) + if "per_tool_values" in stripe: + self.stripe.per_tool_values = { + k: int(v) for k, v in (stripe["per_tool_values"] or {}).items() + } + + gh = data.get("github") or {} + if "token" in gh: + self.github.token = gh["token"] + if "api_url" in gh: + self.github.api_url = gh["api_url"] + if "allowed_repos" in gh: + self.github.allowed_repos = list(gh["allowed_repos"] or []) + if "allow_writes" in gh: + self.github.allow_writes = bool(gh["allow_writes"]) diff --git a/src/superuser/guard.py b/src/superuser/guard.py new file mode 100644 index 0000000..0f03cd1 --- /dev/null +++ b/src/superuser/guard.py @@ -0,0 +1,152 @@ +"""The guard: the single choke point every tool invocation passes through. + +Order of checks for each call: +1. Cisco AI Defense inspects the tool call (may block). +2. Approval gate: risky calls require explicit ``confirm=True`` from a human. +3. The operation runs. +4. Cisco AI Defense inspects the response (may block before it is returned). +5. The outcome is written to the audit log. + +Every branch produces a structured dict so the MCP client always gets a clear, +machine-readable status instead of an opaque error. +""" + +from __future__ import annotations + +import logging +from typing import Any, Callable + +from .audit import AuditLog +from .authz import DuoAuthorizer +from .billing import StripeMeter +from .config import Config, Risk +from .security import CiscoAIDefense + +logger = logging.getLogger("superuser.guard") + + +class Guard: + def __init__( + self, + config: Config, + audit: AuditLog, + defense: CiscoAIDefense, + authorizer: DuoAuthorizer | None = None, + meter: StripeMeter | None = None, + ) -> None: + self.config = config + self.audit = audit + self.defense = defense + self.authorizer = authorizer + self.meter = meter + + def run( + self, + *, + tool_name: str, + args: dict[str, Any], + risk: Risk, + op: Callable[[], Any], + confirm: bool = False, + ) -> dict[str, Any]: + risk_name = risk.name + + # 1. Pre-execution inspection. + pre = self.defense.inspect_tool_call(tool_name, args) + if not pre.safe: + self.audit.record(tool_name, args, decision="blocked_by_ai_defense", + status="denied", risk=risk_name, detail=", ".join(pre.reasons)) + return _denied( + tool_name, + reason="Cisco AI Defense blocked this tool call.", + risk=risk_name, + details=pre.reasons, + ) + + # 2. Duo Agentic Identity authorization (per-tool-call policy). + if self.authorizer is not None: + authz = self.authorizer.authorize(tool_name, args, risk_name) + if not authz.allowed: + self.audit.record(tool_name, args, decision="denied_by_duo", + status="denied", risk=risk_name, detail=authz.reason) + return _denied( + tool_name, + reason="Cisco Duo Agentic Identity denied this tool call.", + risk=risk_name, + details=[authz.reason], + ) + else: + authz = None + + # 3. Approval gate. + if self.config.requires_approval(risk) and not confirm: + self.audit.record(tool_name, args, decision="approval_required", + status="pending", risk=risk_name) + return { + "status": "approval_required", + "tool": tool_name, + "risk": risk_name, + "protected": pre.protected, + "message": ( + f"'{tool_name}' is a {risk_name}-risk action and needs explicit " + "human approval. Review the arguments and call again with confirm=true." + ), + "arguments": args, + } + + # 4. Execute. + try: + output = op() + except Exception as exc: + self.audit.record(tool_name, args, decision="allowed", status="error", + risk=risk_name, detail=str(exc)) + logger.exception("Tool %s failed", tool_name) + return {"status": "error", "tool": tool_name, "error": str(exc)} + + # 5. Post-execution (response) inspection. + response_payload = output if isinstance(output, dict) else {"result": output} + post = self.defense.inspect_response(tool_name, _stringify_for_inspection(response_payload)) + if not post.safe: + self.audit.record(tool_name, args, decision="response_blocked_by_ai_defense", + status="denied", risk=risk_name, detail=", ".join(post.reasons)) + return _denied( + tool_name, + reason="Cisco AI Defense blocked the tool response (possible data leakage).", + risk=risk_name, + details=post.reasons, + ) + + # 6. Meter the successful call (best-effort; never blocks). + billing = self.meter.record(tool_name) if self.meter is not None else None + + # 7. Success audit. + self.audit.record(tool_name, args, decision="allowed", status="success", risk=risk_name) + response: dict[str, Any] = { + "status": "ok", + "tool": tool_name, + "risk": risk_name, + "protected": pre.protected and post.protected, + "result": output, + } + if authz is not None: + response["governed"] = authz.governed + if billing is not None: + response["billed"] = billing.billed + return response + + +def _denied(tool_name: str, *, reason: str, risk: str, details: list[str]) -> dict[str, Any]: + return { + "status": "denied", + "tool": tool_name, + "risk": risk, + "reason": reason, + "details": details, + } + + +def _stringify_for_inspection(payload: dict[str, Any]) -> dict[str, Any]: + """AI Defense response inspection expects text content blocks.""" + import json + + return {"content": [{"type": "text", "text": json.dumps(payload, default=str)[:20000]}]} diff --git a/src/superuser/security.py b/src/superuser/security.py new file mode 100644 index 0000000..8fd068a --- /dev/null +++ b/src/superuser/security.py @@ -0,0 +1,170 @@ +"""Cisco AI Defense integration. + +Two layers of protection are wired in: + +1. Runtime protection via ``agentsec.protect()`` — auto-patches LLM and MCP + client libraries so every model call and tool invocation is inspected. + Enabled once at process start. + +2. Explicit MCP tool-call inspection via ``MCPInspectionClient`` — used by the + guard before a tool runs and after it returns, so this server inspects its + *own* tools even when the caller is not auto-patched. + +The Cisco SDK is imported lazily. If it is not installed, or no credentials are +configured, the server still runs but every tool result is annotated as +UNPROTECTED so the gap is never silent. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass + +from .config import CiscoConfig + +logger = logging.getLogger("superuser.security") + + +@dataclass +class InspectionResult: + """Normalized outcome of an AI Defense inspection.""" + + safe: bool + protected: bool + reasons: list[str] + action: str # "allow", "block", or "unprotected" + + @classmethod + def unprotected(cls) -> "InspectionResult": + return cls(safe=True, protected=False, reasons=[], action="unprotected") + + @classmethod + def allow(cls) -> "InspectionResult": + return cls(safe=True, protected=True, reasons=[], action="allow") + + @classmethod + def block(cls, reasons: list[str]) -> "InspectionResult": + return cls(safe=False, protected=True, reasons=reasons, action="block") + + +class CiscoAIDefense: + """Wraps the Cisco AI Defense SDK with graceful degradation.""" + + def __init__(self, config: CiscoConfig) -> None: + self.config = config + self._mcp_client = None + self._protected = False + self._available = self._probe_sdk() + if config.enabled and not self._available: + logger.warning( + "Cisco AI Defense SDK not installed; tools will run UNPROTECTED. " + "Install with: pip install 'superuser[security]'" + ) + + @staticmethod + def _probe_sdk() -> bool: + try: + import aidefense # noqa: F401 + except Exception: + return False + return True + + def enable_runtime_protection(self) -> bool: + """Call agentsec.protect() to auto-patch LLM + MCP clients. + + Returns True if protection was enabled. + """ + if not (self._available and self.config.can_protect): + return False + try: + from aidefense.runtime import agentsec + + if self.config.agentsec_config_path: + agentsec.protect(config=self.config.agentsec_config_path) + else: + agentsec.protect( + llm_integration_mode="api", + api_mode={ + "llm": { + "mode": self.config.mode, + "endpoint": self.config.runtime_endpoint, + "api_key": self.config.runtime_api_key, + } + }, + ) + self._protected = True + logger.info("Cisco AI Defense runtime protection enabled (mode=%s).", self.config.mode) + except Exception as exc: # protection must never crash the server + logger.error("Failed to enable Cisco AI Defense runtime protection: %s", exc) + return self._protected + + def _client(self): + if self._mcp_client is not None: + return self._mcp_client + if not (self._available and self.config.can_inspect): + return None + try: + from aidefense import MCPInspectionClient + + self._mcp_client = MCPInspectionClient(api_key=self.config.inspection_api_key) + except Exception as exc: + logger.error("Could not initialize MCPInspectionClient: %s", exc) + self._mcp_client = None + return self._mcp_client + + def inspect_tool_call(self, tool_name: str, arguments: dict) -> InspectionResult: + client = self._client() + if client is None: + return InspectionResult.unprotected() + try: + result = client.inspect_tool_call(tool_name=tool_name, arguments=arguments) + return self._normalize(result) + except Exception as exc: + logger.error("AI Defense tool-call inspection error: %s", exc) + # Fail closed in enforce mode, open in monitor mode. + if self.config.mode == "enforce": + return InspectionResult.block([f"inspection error: {exc}"]) + return InspectionResult.unprotected() + + def inspect_response(self, tool_name: str, result_data: dict) -> InspectionResult: + client = self._client() + if client is None: + return InspectionResult.unprotected() + try: + result = client.inspect_response( + result_data=result_data, + method="tools/call", + params={"name": tool_name}, + ) + return self._normalize(result) + except Exception as exc: + logger.error("AI Defense response inspection error: %s", exc) + if self.config.mode == "enforce": + return InspectionResult.block([f"inspection error: {exc}"]) + return InspectionResult.unprotected() + + def _normalize(self, result) -> InspectionResult: + """Map the SDK's response object onto our InspectionResult.""" + inner = getattr(result, "result", result) + is_safe = bool(getattr(inner, "is_safe", True)) + reasons: list[str] = [] + for rule in getattr(inner, "rules", None) or []: + name = getattr(rule, "rule_name", None) or getattr(rule, "classification", "rule") + reasons.append(str(name)) + if is_safe: + return InspectionResult.allow() + # In monitor mode we surface the finding but do not block. + if self.config.mode == "monitor": + logger.warning("AI Defense flagged (monitor): %s", ", ".join(reasons) or "unsafe") + return InspectionResult(safe=True, protected=True, reasons=reasons, action="allow") + return InspectionResult.block(reasons or ["policy violation"]) + + @property + def status(self) -> dict: + return { + "sdk_installed": self._available, + "enabled": self.config.enabled, + "mode": self.config.mode, + "runtime_protection": self._protected, + "inspection_configured": bool(self.config.can_inspect), + } diff --git a/src/superuser/server.py b/src/superuser/server.py new file mode 100644 index 0000000..f26ba93 --- /dev/null +++ b/src/superuser/server.py @@ -0,0 +1,138 @@ +"""MCP server exposing the superuser toolset behind the guard.""" + +from __future__ import annotations + +import logging +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from .audit import AuditLog +from .authz import DuoAuthorizer +from .billing import StripeMeter +from .config import Config, Risk +from .guard import Guard +from .security import CiscoAIDefense +from .tools import github as gh_tools +from .tools import shell as shell_tools +from .tools import system as system_tools + +logger = logging.getLogger("superuser.server") + + +def build_server(config: Config | None = None) -> FastMCP: + config = config or Config.load() + audit = AuditLog(config.audit_log_path, human_owner=config.human_owner) + defense = CiscoAIDefense(config.cisco) + defense.enable_runtime_protection() + authorizer = DuoAuthorizer(config.duo) + meter = StripeMeter(config.stripe) + guard = Guard(config, audit, defense, authorizer=authorizer, meter=meter) + github_client = gh_tools.GitHubClient(config.github) + + mcp = FastMCP("superuser") + + # -- System inspection (low risk, read-only) -------------------------------- + + @mcp.tool() + def system_status() -> dict[str, Any]: + """Report host, CPU, memory, and uptime for this machine.""" + return guard.run( + tool_name="system_status", args={}, risk=Risk.LOW, + op=system_tools.system_status, + ) + + @mcp.tool() + def disk_usage() -> dict[str, Any]: + """Report disk usage per mounted partition.""" + return guard.run( + tool_name="disk_usage", args={}, risk=Risk.LOW, op=system_tools.disk_usage, + ) + + @mcp.tool() + def list_processes(limit: int = 20, sort_by: str = "memory") -> dict[str, Any]: + """List top processes. sort_by is 'memory' or 'cpu'.""" + return guard.run( + tool_name="list_processes", + args={"limit": limit, "sort_by": sort_by}, + risk=Risk.LOW, + op=lambda: system_tools.list_processes(limit=limit, sort_by=sort_by), + ) + + # -- Shell execution (high risk, approval-gated) ---------------------------- + + @mcp.tool() + def run_command(command: str, timeout: int = 0, confirm: bool = False) -> dict[str, Any]: + """Run a shell command on this machine. + + Non-allowlisted commands are HIGH risk and require confirm=true. Commands + matching the dangerous-command denylist are always rejected, and every + call is inspected by Cisco AI Defense and written to the audit log. + """ + effective_timeout = timeout or config.command_timeout_seconds + return guard.run( + tool_name="run_command", + args={"command": command, "timeout": effective_timeout}, + risk=shell_tools.classify_risk(command, config), + confirm=confirm, + op=lambda: shell_tools.execute_command(command, config, effective_timeout), + ) + + # -- GitHub (least-privilege) ----------------------------------------------- + + @mcp.tool() + def github_get_repo(repo: str) -> dict[str, Any]: + """Get metadata for an allowlisted repo (e.g. 'owner/name').""" + return guard.run( + tool_name="github_get_repo", args={"repo": repo}, risk=Risk.LOW, + op=lambda: github_client.get_repo(repo), + ) + + @mcp.tool() + def github_list_issues(repo: str, state: str = "open", limit: int = 20) -> dict[str, Any]: + """List issues for an allowlisted repo.""" + return guard.run( + tool_name="github_list_issues", + args={"repo": repo, "state": state, "limit": limit}, + risk=Risk.LOW, + op=lambda: github_client.list_issues(repo, state=state, limit=limit), + ) + + @mcp.tool() + def github_create_issue_comment( + repo: str, issue_number: int, body: str, confirm: bool = False + ) -> dict[str, Any]: + """Comment on an issue/PR. Write action: requires confirm=true and GITHUB_ALLOW_WRITES.""" + return guard.run( + tool_name="github_create_issue_comment", + args={"repo": repo, "issue_number": issue_number, "body": body}, + risk=Risk.HIGH, + confirm=confirm, + op=lambda: github_client.create_issue_comment(repo, issue_number, body), + ) + + # -- Introspection ----------------------------------------------------------- + + @mcp.tool() + def security_status() -> dict[str, Any]: + """Report the agent's guardrail configuration and Cisco AI Defense status.""" + return { + "status": "ok", + "result": { + "human_owner": config.human_owner, + "approval_required_at": config.approval_required_at.name, + "command_allowlist": config.command_allowlist, + "denylist_entries": len(config.command_denylist), + "audit_log": str(config.audit_log_path), + "github": { + "configured": bool(config.github.token), + "allowed_repos": config.github.allowed_repos, + "writes_allowed": config.github.allow_writes, + }, + "cisco_ai_defense": defense.status, + "cisco_duo_agentic_identity": authorizer.status, + "stripe_billing": meter.status, + }, + } + + return mcp diff --git a/src/superuser/tools/__init__.py b/src/superuser/tools/__init__.py new file mode 100644 index 0000000..0242ac1 --- /dev/null +++ b/src/superuser/tools/__init__.py @@ -0,0 +1 @@ +"""Tool implementations for the superuser agent.""" diff --git a/src/superuser/tools/github.py b/src/superuser/tools/github.py new file mode 100644 index 0000000..0dc63ed --- /dev/null +++ b/src/superuser/tools/github.py @@ -0,0 +1,90 @@ +"""Least-privilege GitHub access. + +Uses a fine-grained token supplied via config/env. Reads are restricted to an +explicit repo allowlist; writes additionally require ``allow_writes`` to be on +*and* pass the guard's approval gate. The client never falls back to ambient +credentials, so an unconfigured agent simply cannot reach GitHub. +""" + +from __future__ import annotations + +from typing import Any + +import requests + +from ..config import GitHubConfig + + +class GitHubError(Exception): + pass + + +class GitHubClient: + def __init__(self, config: GitHubConfig) -> None: + self.config = config + + def _headers(self) -> dict[str, str]: + if not self.config.token: + raise GitHubError( + "No GitHub token configured. Set GITHUB_TOKEN to a fine-grained, " + "least-privilege token scoped to the repositories you allow." + ) + return { + "Authorization": f"Bearer {self.config.token}", + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + } + + def _check_repo(self, repo: str) -> None: + if not self.config.repo_allowed(repo): + raise GitHubError( + f"Repository {repo!r} is not in the allowlist. " + "Add it to GITHUB_ALLOWED_REPOS to permit access." + ) + + def _request(self, method: str, path: str, **kwargs: Any) -> Any: + url = f"{self.config.api_url}{path}" + resp = requests.request(method, url, headers=self._headers(), timeout=30, **kwargs) + if resp.status_code >= 400: + raise GitHubError(f"GitHub API {resp.status_code}: {resp.text[:500]}") + if resp.text: + return resp.json() + return {} + + def get_repo(self, repo: str) -> dict[str, Any]: + self._check_repo(repo) + data = self._request("GET", f"/repos/{repo}") + return { + "full_name": data.get("full_name"), + "private": data.get("private"), + "default_branch": data.get("default_branch"), + "open_issues": data.get("open_issues_count"), + "pushed_at": data.get("pushed_at"), + "permissions": data.get("permissions"), + } + + def list_issues(self, repo: str, state: str = "open", limit: int = 20) -> dict[str, Any]: + self._check_repo(repo) + items = self._request( + "GET", f"/repos/{repo}/issues", + params={"state": state, "per_page": max(1, min(limit, 100))}, + ) + return { + "repo": repo, + "issues": [ + {"number": i["number"], "title": i["title"], "state": i["state"], + "user": i["user"]["login"], "url": i["html_url"]} + for i in items if "pull_request" not in i + ], + } + + def create_issue_comment(self, repo: str, issue_number: int, body: str) -> dict[str, Any]: + self._check_repo(repo) + if not self.config.allow_writes: + raise GitHubError( + "GitHub writes are disabled. Set GITHUB_ALLOW_WRITES=true to permit them." + ) + data = self._request( + "POST", f"/repos/{repo}/issues/{issue_number}/comments", json={"body": body} + ) + return {"id": data.get("id"), "url": data.get("html_url")} diff --git a/src/superuser/tools/shell.py b/src/superuser/tools/shell.py new file mode 100644 index 0000000..a1dbdee --- /dev/null +++ b/src/superuser/tools/shell.py @@ -0,0 +1,61 @@ +"""Shell command execution with denylist and allowlist controls.""" + +from __future__ import annotations + +import shlex +import subprocess +from typing import Any + +from ..config import Config, Risk + + +class CommandRejected(Exception): + """Raised when a command is blocked by policy before execution.""" + + +def _matches_denylist(command: str, denylist: list[str]) -> str | None: + normalized = " ".join(command.split()).lower() + for pattern in denylist: + if pattern.lower() in normalized: + return pattern + return None + + +def _base_command(command: str) -> str: + try: + parts = shlex.split(command) + except ValueError: + parts = command.split() + return parts[0] if parts else "" + + +def classify_risk(command: str, config: Config) -> Risk: + """Allowlisted base commands are LOW risk; everything else is HIGH.""" + if _base_command(command) in config.command_allowlist: + return Risk.LOW + return Risk.HIGH + + +def execute_command(command: str, config: Config, timeout: int | None = None) -> dict[str, Any]: + hit = _matches_denylist(command, config.command_denylist) + if hit is not None: + raise CommandRejected(f"Command matches denylist entry: {hit!r}") + + timeout = timeout or config.command_timeout_seconds + try: + completed = subprocess.run( + command, + shell=True, + capture_output=True, + text=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired: + raise CommandRejected(f"Command timed out after {timeout}s") + + return { + "command": command, + "exit_code": completed.returncode, + "stdout": completed.stdout[-20000:], + "stderr": completed.stderr[-20000:], + } diff --git a/src/superuser/tools/system.py b/src/superuser/tools/system.py new file mode 100644 index 0000000..fb14705 --- /dev/null +++ b/src/superuser/tools/system.py @@ -0,0 +1,69 @@ +"""Read-only system inspection helpers (low risk).""" + +from __future__ import annotations + +import getpass +import platform +import socket +from datetime import datetime, timezone +from typing import Any + +import psutil + + +def system_status() -> dict[str, Any]: + vm = psutil.virtual_memory() + boot = datetime.fromtimestamp(psutil.boot_time(), tz=timezone.utc) + load_avg: tuple[float, float, float] | None + try: + load_avg = psutil.getloadavg() + except (AttributeError, OSError): + load_avg = None + + return { + "hostname": socket.gethostname(), + "user": getpass.getuser(), + "platform": platform.platform(), + "python": platform.python_version(), + "cpu_count": psutil.cpu_count(logical=True), + "cpu_percent": psutil.cpu_percent(interval=0.2), + "load_average": load_avg, + "memory": { + "total_mb": round(vm.total / 1_048_576, 1), + "available_mb": round(vm.available / 1_048_576, 1), + "percent_used": vm.percent, + }, + "boot_time_utc": boot.isoformat(), + } + + +def disk_usage() -> dict[str, Any]: + partitions = [] + for part in psutil.disk_partitions(all=False): + try: + usage = psutil.disk_usage(part.mountpoint) + except PermissionError: + continue + partitions.append({ + "device": part.device, + "mountpoint": part.mountpoint, + "fstype": part.fstype, + "total_gb": round(usage.total / 1_073_741_824, 2), + "used_gb": round(usage.used / 1_073_741_824, 2), + "percent_used": usage.percent, + }) + return {"partitions": partitions} + + +def list_processes(limit: int = 20, sort_by: str = "memory") -> dict[str, Any]: + limit = max(1, min(limit, 200)) + procs = [] + for proc in psutil.process_iter(["pid", "name", "username", "cpu_percent", "memory_percent"]): + try: + procs.append(proc.info) + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + + key = "cpu_percent" if sort_by == "cpu" else "memory_percent" + procs.sort(key=lambda p: p.get(key) or 0.0, reverse=True) + return {"count": len(procs), "sorted_by": sort_by, "processes": procs[:limit]} diff --git a/superuser.example.yaml b/superuser.example.yaml new file mode 100644 index 0000000..cd5b55c --- /dev/null +++ b/superuser.example.yaml @@ -0,0 +1,75 @@ +# Superuser agent configuration. Copy to a file and pass with +# `superuser --config ` or set SUPERUSER_CONFIG. + +# Human accountable for this agent's actions (recorded in every audit entry). +human_owner: "jane.doe@example.com" + +# Minimum risk level that triggers the human-approval gate. +# One of: LOW, MEDIUM, HIGH, CRITICAL. +approval_required_at: MEDIUM + +# Base commands that are treated as LOW risk (no approval needed). +command_allowlist: + - ls + - cat + - df + - uptime + - whoami + - uname + +# Extra dangerous patterns to reject (merged with the built-in denylist). +command_denylist: + - "rm -rf /*" + +command_timeout_seconds: 60 +audit_log_path: "superuser-audit.log" + +cisco: + enabled: true + mode: enforce # enforce | monitor | off + # Prefer setting secrets via environment variables, not in this file: + # AI_DEFENSE_INSPECTION_API_KEY + # AI_DEFENSE_API_MODE_LLM_ENDPOINT / AI_DEFENSE_API_MODE_LLM_API_KEY + # agentsec_config_path: "agentsec.yaml" + +# Cisco Duo Agentic Identity: per-tool-call authorization (OAuth 2.1). +# "Duo decides, the gateway enforces." Prefer env vars for client secrets. +duo: + enabled: false + mode: enforce # enforce | monitor | off + agent_id: "superuser-agent-1" + # token_url: "https://.duosecurity.com/oauth/v1/token" + # client_id: "..." # prefer DUO_CLIENT_ID env var + # client_secret: "..." # prefer DUO_CLIENT_SECRET env var + # resource: "https://superuser.internal" # RFC 8707 audience + # authorization_url: "https://.../authorize" # optional decision endpoint + scopes: + - superuser.read + - superuser.exec + # Scope required to invoke each tool (scope-based access control): + tool_scopes: + system_status: superuser.read + disk_usage: superuser.read + list_processes: superuser.read + github_get_repo: superuser.read + github_list_issues: superuser.read + run_command: superuser.exec + github_create_issue_comment: superuser.write + +# Stripe usage-based billing: meter each successful tool call. +stripe: + enabled: false + # api_key: "sk_live_..." # prefer STRIPE_API_KEY env var + # customer_id: "cus_..." # prefer STRIPE_CUSTOMER_ID env var + event_name: superuser_tool_call + default_value: 1 + # Charge more for higher-risk actions: + per_tool_values: + run_command: 5 + github_create_issue_comment: 3 + +github: + # Prefer GITHUB_TOKEN env var over putting the token here. + allowed_repos: + - "your-org/your-repo" + allow_writes: false diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_guard.py b/tests/test_guard.py new file mode 100644 index 0000000..224d3fe --- /dev/null +++ b/tests/test_guard.py @@ -0,0 +1,181 @@ +"""Tests for the guard, config, shell policy, and audit log.""" + +from __future__ import annotations + +import json + +import pytest + +from superuser.audit import AuditLog +from superuser.authz import AuthzDecision, DuoAuthorizer +from superuser.billing import StripeMeter +from superuser.config import Config, DuoConfig, Risk, StripeConfig +from superuser.guard import Guard +from superuser.security import CiscoAIDefense +from superuser.tools import shell as shell_tools + + +@pytest.fixture +def guard(tmp_path): + config = Config() + config.approval_required_at = Risk.MEDIUM + config.audit_log_path = tmp_path / "audit.log" + # Cisco disabled => unprotected but non-blocking, so we test the local gate. + config.cisco.enabled = False + audit = AuditLog(config.audit_log_path, human_owner="tester") + defense = CiscoAIDefense(config.cisco) + authorizer = DuoAuthorizer(config.duo) + meter = StripeMeter(config.stripe) + return Guard(config, audit, defense, authorizer, meter), config, audit + + +def test_low_risk_runs_without_confirmation(guard): + g, _, _ = guard + result = g.run(tool_name="probe", args={}, risk=Risk.LOW, op=lambda: {"ok": True}) + assert result["status"] == "ok" + assert result["result"] == {"ok": True} + + +def test_high_risk_requires_confirmation(guard): + g, _, _ = guard + ran = {"called": False} + + def op(): + ran["called"] = True + return "done" + + result = g.run(tool_name="danger", args={"x": 1}, risk=Risk.HIGH, op=op) + assert result["status"] == "approval_required" + assert ran["called"] is False + + +def test_high_risk_runs_with_confirmation(guard): + g, _, _ = guard + result = g.run(tool_name="danger", args={}, risk=Risk.HIGH, confirm=True, op=lambda: "done") + assert result["status"] == "ok" + assert result["result"] == "done" + + +def test_errors_are_captured_and_audited(guard): + g, _, audit = guard + + def boom(): + raise RuntimeError("kaboom") + + result = g.run(tool_name="boom", args={}, risk=Risk.LOW, op=boom) + assert result["status"] == "error" + assert "kaboom" in result["error"] + lines = audit.path.read_text().strip().splitlines() + assert any(json.loads(line)["status"] == "error" for line in lines) + + +def test_audit_redacts_secrets(tmp_path): + audit = AuditLog(tmp_path / "a.log", human_owner="tester") + audit.record("t", {"token": "supersecret", "cmd": "ls"}, "allowed", "success") + entry = json.loads(audit.path.read_text().strip()) + assert entry["arguments"]["token"] == "***REDACTED***" + assert entry["arguments"]["cmd"] == "ls" + + +def test_denylist_blocks_dangerous_commands(): + config = Config() + with pytest.raises(shell_tools.CommandRejected): + shell_tools.execute_command("rm -rf /", config) + + +def test_allowlist_classifies_low_risk(): + config = Config() + config.command_allowlist = ["ls"] + assert shell_tools.classify_risk("ls -la", config) == Risk.LOW + assert shell_tools.classify_risk("curl http://x", config) == Risk.HIGH + + +def test_command_executes_and_captures_output(): + config = Config() + result = shell_tools.execute_command("echo hello", config) + assert result["exit_code"] == 0 + assert "hello" in result["stdout"] + + +def test_github_repo_allowlist_fails_closed(): + from superuser.config import GitHubConfig + from superuser.tools.github import GitHubClient, GitHubError + + client = GitHubClient(GitHubConfig(token="x", allowed_repos=[])) + with pytest.raises(GitHubError): + client.get_repo("owner/repo") + + +# --- Cisco Duo authorization ------------------------------------------------ + +def test_duo_ungoverned_when_not_configured(): + authz = DuoAuthorizer(DuoConfig(enabled=False)) + decision = authz.authorize("run_command", {}, "HIGH") + assert decision.allowed is True + assert decision.governed is False + + +def test_duo_scope_based_allow_and_deny(): + cfg = DuoConfig( + enabled=True, mode="enforce", token_url="https://duo.example/token", + client_id="c", client_secret="s", + tool_scopes={"run_command": "superuser.exec"}, + ) + authz = DuoAuthorizer(cfg) + # Pretend Duo issued a token granting only the read scope. + authz._token = "tok" + authz._token_expiry = float("inf") + authz._granted_scopes = {"superuser.read"} + + assert authz._authorize_via_scopes("run_command").allowed is False + authz._granted_scopes = {"superuser.exec"} + assert authz._authorize_via_scopes("run_command").allowed is True + + +def test_duo_denial_blocks_and_audits(guard, monkeypatch): + g, _, audit = guard + monkeypatch.setattr( + g.authorizer, "authorize", + lambda *a, **k: AuthzDecision.deny("missing scope"), + ) + ran = {"called": False} + + def op(): + ran["called"] = True + return "done" + + result = g.run(tool_name="run_command", args={}, risk=Risk.HIGH, confirm=True, op=op) + assert result["status"] == "denied" + assert ran["called"] is False + lines = audit.path.read_text().strip().splitlines() + assert any(json.loads(line)["decision"] == "denied_by_duo" for line in lines) + + +def test_duo_governed_flag_on_success(guard, monkeypatch): + g, _, _ = guard + monkeypatch.setattr( + g.authorizer, "authorize", lambda *a, **k: AuthzDecision.allow("ok"), + ) + result = g.run(tool_name="system_status", args={}, risk=Risk.LOW, op=lambda: {"ok": True}) + assert result["status"] == "ok" + assert result["governed"] is True + + +# --- Stripe billing --------------------------------------------------------- + +def test_billing_disabled_by_default(): + meter = StripeMeter(StripeConfig(enabled=False)) + assert meter.record("system_status").billed is False + + +def test_billing_per_tool_value(): + cfg = StripeConfig(enabled=True, per_tool_values={"run_command": 5}, default_value=1) + assert cfg.value_for("run_command") == 5 + assert cfg.value_for("system_status") == 1 + + +def test_billed_flag_present_on_success(guard): + g, _, _ = guard + result = g.run(tool_name="system_status", args={}, risk=Risk.LOW, op=lambda: {"ok": True}) + assert result["status"] == "ok" + assert result["billed"] is False # billing disabled in fixture