diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..9e77b7b --- /dev/null +++ b/.dockerignore @@ -0,0 +1,16 @@ +.git +.github +.venv +.intentgate-state +__pycache__ +*.py[cod] +*.egg-info +.env +.env.* +!.env.integrations.example +config/integrations.json +tests +docs +observability +deploy +*.log diff --git a/.env.integrations.example b/.env.integrations.example new file mode 100644 index 0000000..ea3f481 --- /dev/null +++ b/.env.integrations.example @@ -0,0 +1,19 @@ +# Copy to .env.integrations and populate only integrations you enable. +DEFENDER_XDR_TOKEN= +MICROSOFT_SENTINEL_TOKEN= +CROWDSTRIKE_ACCESS_TOKEN= +SENTINELONE_API_TOKEN= +SPLUNK_TOKEN= +ELASTIC_API_KEY= + +# Change these before exposing the stack beyond localhost. +UIG_INGEST_TOKEN=change-me +GRAFANA_ADMIN_USER=admin +GRAFANA_ADMIN_PASSWORD=change-me + +# Manager escalation webhook (generic JSON, Slack, or Teams). +UIG_MANAGER_ID= +UIG_MANAGER_REPORT_THRESHOLD=70 +UIG_MANAGER_WEBHOOK_URL= +UIG_MANAGER_WEBHOOK_TOKEN= +UIG_MANAGER_WEBHOOK_STYLE=generic diff --git a/.github/ISSUE_TEMPLATE/bug_report.yml b/.github/ISSUE_TEMPLATE/bug_report.yml new file mode 100644 index 0000000..e42e9c7 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/bug_report.yml @@ -0,0 +1,37 @@ +name: Bug report +description: Report a reproducible defect or incorrect policy outcome. +title: "[Bug]: " +labels: [bug] +body: + - type: markdown + attributes: + value: Do not include credentials, private command history, or sensitive security events. + - type: textarea + id: description + attributes: + label: Description + description: What happened, and what did you expect? + validations: + required: true + - type: textarea + id: reproduction + attributes: + label: Reproduction + description: Provide a minimal sanitized command and configuration. + validations: + required: true + - type: input + id: environment + attributes: + label: Environment + description: OS, Python version, shell, and project version. + validations: + required: true + - type: dropdown + id: decision + attributes: + label: Incorrect decision + options: [ALLOW, REVIEW, BLOCK, Not applicable] + validations: + required: true + diff --git a/.github/ISSUE_TEMPLATE/feature_request.yml b/.github/ISSUE_TEMPLATE/feature_request.yml new file mode 100644 index 0000000..edff9f5 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/feature_request.yml @@ -0,0 +1,26 @@ +name: Feature request +description: Propose a policy, integration, workflow, or observability improvement. +title: "[Feature]: " +labels: [enhancement] +body: + - type: textarea + id: problem + attributes: + label: Problem + description: What security or usability problem should this solve? + validations: + required: true + - type: textarea + id: proposal + attributes: + label: Proposed approach + description: Describe the behavior, relevant platforms, and expected decision impact. + validations: + required: true + - type: textarea + id: tradeoffs + attributes: + label: False-positive, privacy, and latency tradeoffs + validations: + required: true + diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..520f96e --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,15 @@ +version: 2 +updates: + - package-ecosystem: pip + directory: "/" + schedule: + interval: monthly + - package-ecosystem: github-actions + directory: "/" + schedule: + interval: monthly + - package-ecosystem: docker + directory: "/" + schedule: + interval: monthly + diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..d895093 --- /dev/null +++ b/.github/pull_request_template.md @@ -0,0 +1,18 @@ +## Summary + +Describe what changed and why. + +## Security impact + +- Decision or scoring changes: +- False-positive / false-negative tradeoffs: +- Privacy or telemetry changes: + +## Validation + +- [ ] Unit tests pass +- [ ] New behavior includes tests +- [ ] Python sources compile +- [ ] Docker Compose validates when affected +- [ ] No credentials or private telemetry are included + diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..1a22478 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,69 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +permissions: + contents: read + +jobs: + test: + strategy: + matrix: + os: [ubuntu-latest, windows-latest] + python-version: ["3.11", "3.13"] + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + - name: Install + run: python -m pip install -e . + - name: Test + run: python -m unittest discover -s tests -v + - name: Compile + run: python -m compileall -q src tests + + compose: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Validate Docker Compose + run: docker compose -f docker-compose.observability.yml config --quiet + + terraform: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: hashicorp/setup-terraform@v3 + with: + terraform_version: "1.14.0" + - name: Check formatting + run: terraform -chdir=deploy/terraform fmt -check -recursive + - name: Initialize providers + run: terraform -chdir=deploy/terraform init -backend=false + - name: Validate configuration + run: terraform -chdir=deploy/terraform validate + + ansible: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: pip + - name: Install Ansible + run: python -m pip install "ansible-core>=2.18,<2.20" ansible-lint + - name: Install collections + run: ansible-galaxy collection install -r deploy/ansible/requirements.yml + - name: Syntax check + working-directory: deploy/ansible + run: ansible-playbook -i inventories/example.ini deploy.yml --syntax-check + - name: Lint playbook and role + working-directory: deploy/ansible + run: ansible-lint deploy.yml roles diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..29c364e --- /dev/null +++ b/.gitignore @@ -0,0 +1,37 @@ +.venv/ +__pycache__/ +*.py[cod] +.pytest_cache/ +.test-state/ +.intentgate-state/ +dist/ +build/ +*.egg-info/ +.env +.env.* +!.env.integrations.example +config/integrations.json +*.log +*.local.* +history.jsonl +manager-reports/ + +# Terraform state and local inputs +**/.terraform/ +*.tfstate +*.tfstate.* +*.tfvars +*.tfvars.json +!*.tfvars.example +crash.log +override.tf +override.tf.json +*_override.tf +*_override.tf.json + +# Ansible operator inventory, Vault data, and retry artifacts +deploy/ansible/inventories/*.ini +!deploy/ansible/inventories/example.ini +deploy/ansible/group_vars/all.yml +deploy/ansible/host_vars/ +*.retry diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..4b129a1 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,21 @@ +# Changelog + +All notable changes to this project will be documented here. + +## [0.3.0] - 2026-08-12 + +### Added + +- Low-latency `ALLOW`, `REVIEW`, and `BLOCK` command policy engine +- Windows and Linux privilege detection +- Cross-platform destructive-action catalog +- Per-user behavioral command baseline and anomaly scoring +- Cached project provenance and code-health scanning +- Normalized AV, EDR, and SIEM signal ingestion +- Microsoft Defender local collector and enterprise integration templates +- Prometheus metrics and provisioned Grafana dashboard +- Redacted asynchronous manager/security risk reports +- Generic, Slack, and Microsoft Teams webhook delivery +- Automated tests and repository governance documentation +- Terraform Docker Compose deployment module +- Ansible Linux host bootstrap and secure deployment role diff --git a/CODE_OF_CONDUCT.md b/CODE_OF_CONDUCT.md new file mode 100644 index 0000000..a621cbf --- /dev/null +++ b/CODE_OF_CONDUCT.md @@ -0,0 +1,22 @@ +# Code of Conduct + +## Our commitment + +We are committed to a welcoming, respectful, and harassment-free community for everyone, regardless of background, identity, experience, or viewpoint. + +## Expected behavior + +- Be constructive, specific, and respectful. +- Assume good faith while challenging ideas with evidence. +- Respect privacy and never post credentials, private telemetry, or identifying command history. +- Accept responsibility, apologize when appropriate, and learn from mistakes. +- Keep security research safe, lawful, and proportionate. + +## Unacceptable behavior + +Harassment, threats, discrimination, doxxing, deliberate disruption, sexualized conduct, and publishing another person's private information are not acceptable. + +## Enforcement + +Repository maintainers may edit, remove, or reject contributions and temporarily or permanently restrict participation when behavior violates these expectations. Report conduct concerns privately to the repository owner through an appropriate GitHub channel. + diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..0227874 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,45 @@ +# Contributing + +Thank you for helping improve User Intent AI Security. + +## Development setup + +```powershell +python -m venv .venv +.\.venv\Scripts\python.exe -m pip install -e . +.\.venv\Scripts\python.exe -m unittest discover -s tests -v +``` + +Linux and macOS users can substitute `.venv/bin/python`. + +## Pull requests + +1. Open an issue first for major policy, architecture, telemetry, or privacy changes. +2. Keep each pull request focused on one coherent outcome. +3. Add tests for new rules, integrations, redaction behavior, and decision changes. +4. Explain false-positive and false-negative tradeoffs for security detections. +5. Never commit real credentials, customer events, employee command history, or private manager reports. +6. Run the test suite, bytecode compilation, and Compose validation before requesting review. + +```powershell +.\.venv\Scripts\python.exe -m unittest discover -s tests -v +.\.venv\Scripts\python.exe -m compileall -q src tests +docker compose -f docker-compose.observability.yml config --quiet +``` + +## Detection contributions + +New destructive-action patterns should include: + +- The platforms and command families affected +- A concise risk category and explanation +- A conservative score justified by likely impact +- Positive and negative tests +- Consideration of quoting, aliases, mixed case, and benign administrative usage + +Avoid broad patterns that classify ordinary read-only commands as destructive. + +## Integration contributions + +Use the normalized signal model. Keep credentials in environment variables, bound network operations, use event IDs for deduplication, define TTL behavior, and document the vendor API version used. + diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..9edbadf --- /dev/null +++ b/Dockerfile @@ -0,0 +1,8 @@ +FROM python:3.13-slim +WORKDIR /app +COPY pyproject.toml README.md ./ +COPY src ./src +RUN pip install --no-cache-dir . +EXPOSE 8787 +CMD ["uig-service", "--host", "0.0.0.0", "--port", "8787"] + diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..b8a446d --- /dev/null +++ b/LICENSE @@ -0,0 +1,22 @@ +MIT License + +Copyright (c) 2026 BB AI Arena + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + diff --git a/README.md b/README.md index 006079f..e4ef6ca 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,275 @@ -# User-Intent-AI-Security -Wrapper to determine the users intent before allowing execution. +
+ User Intent AI Security +
+ +
+ +[![CI](https://github.com/BB-AI-Arena/User-Intent-AI-Security/actions/workflows/ci.yml/badge.svg)](https://github.com/BB-AI-Arena/User-Intent-AI-Security/actions/workflows/ci.yml) +[![Python](https://img.shields.io/badge/Python-3.11%2B-3776AB?logo=python&logoColor=white)](https://www.python.org/) +[![License: MIT](https://img.shields.io/badge/License-MIT-2ea44f.svg)](LICENSE) +[![Status: POC](https://img.shields.io/badge/status-proof%20of%20concept-f59e0b)](#project-status) +[![Docker](https://img.shields.io/badge/observability-Docker%20Compose-2496ED?logo=docker&logoColor=white)](docker-compose.observability.yml) +[![Terraform](https://img.shields.io/badge/IaC-Terraform-844FBA?logo=terraform&logoColor=white)](deploy/terraform) +[![Ansible](https://img.shields.io/badge/automation-Ansible-EE0000?logo=ansible&logoColor=white)](deploy/ansible) + +**A context-aware command execution gate that asks one critical question before code runs: _does this action match the user's intent?_** + +[Quick start](#quick-start) · [How it works](#how-it-works) · [Integrations](#security-integrations) · [Dashboard](#grafana-dashboard) · [Security](SECURITY.md) + +
+ +> [!IMPORTANT] +> User Intent AI Security is an early proof of concept, not a standalone security boundary. Reliable enforcement requires integration at a command broker, shell host, agent tool API, container runtime, or privileged-operation choke point that cannot be bypassed. + +## Why this project exists + +Traditional endpoint controls ask whether a command is known to be malicious. That is necessary, but incomplete. A legitimate administrative command can still be dangerous when it is unexpected, mistyped, overly broad, generated without adequate review, executed under elevated privileges, or inconsistent with the task the user was performing. + +User Intent AI Security adds a pre-execution decision layer. It combines command semantics with live context and returns one of three outcomes: + +| Decision | Meaning | Default behavior | +|---|---|---| +| `ALLOW` | Low-risk and consistent with available context | Execute silently | +| `REVIEW` | Ambiguous, elevated, or externally consequential | Require confirmation | +| `BLOCK` | High-confidence destructive or suspicious behavior | Do not execute | + +The ordinary-command hot path is local and deterministic. External APIs, SIEM queries, provenance scans, and manager notifications run asynchronously and contribute cached signals without delaying execution. + +## Highlights + +- **Sub-millisecond policy evaluation** — approximately 0.53 ms for context collection and scoring in the development benchmark. +- **Intent-aware decisions** — evaluates declared purpose, recent activity, Git state, command behavior, and projected blast radius. +- **Privilege-sensitive policy** — detects Linux root, Linux administrative groups, and elevated Windows administrator sessions. +- **Cross-platform destructive-action catalog** — covers filesystem, storage, identity, networking, services, databases, containers, cloud resources, recovery controls, and audit-log changes. +- **Behavioral anomaly detection** — learns a local per-user baseline and identifies unusual command families. +- **AV, EDR, and SIEM context** — accepts normalized signals from common security platforms and correlates them by confidence and scope. +- **AI-assisted code risk signals** — performs a bounded, cached code-health/provenance scan without claiming unreliable authorship detection. +- **Privacy-conscious escalation** — queues redacted risk reports for an approved manager or security webhook. +- **Operational visibility** — ships with Prometheus metrics and a provisioned Grafana dashboard. +- **Standard-library core** — the gate has no mandatory runtime dependencies outside Python 3.11+. + +## How it works + +```mermaid +flowchart LR + U["User or agent"] --> G["Intent Gate"] + G --> C["Bounded local context"] + C --> P["Policy engine"] + AV["AV / EDR / SIEM"] --> N["Normalized signal cache"] + S["Provenance scanner"] --> N + N --> P + P -->|"Low risk"| A["ALLOW"] + P -->|"Ambiguous"| R["REVIEW"] + P -->|"High risk"| B["BLOCK"] + P --> Q["Audit + redacted report queue"] + Q --> M["Manager / security webhook"] + Q --> O["Prometheus + Grafana"] +``` + +The decision engine considers: + +1. The command and declared purpose. +2. Destructive capabilities and target scope. +3. Current user privilege and repository state. +4. Recent commands and the user's learned baseline. +5. Cached project provenance and code-health indicators. +6. Non-expired AV, EDR, and SIEM posture signals scoped to the device, user, or project. + +See [Architecture](docs/ARCHITECTURE.md) and [Threat Model](docs/THREAT_MODEL.md) for the deeper design. + +## Quick start + +### Windows PowerShell + +```powershell +git clone https://github.com/BB-AI-Arena/User-Intent-AI-Security.git +cd User-Intent-AI-Security + +python -m venv .venv +.\.venv\Scripts\python.exe -m pip install -e . +. .\scripts\Enable-IntentGate.ps1 + +uig -- git status +uig --purpose "publish reviewed changes" --explain --dry-run -- git push +``` + +### Linux or macOS + +```bash +git clone https://github.com/BB-AI-Arena/User-Intent-AI-Security.git +cd User-Intent-AI-Security + +python3 -m venv .venv +.venv/bin/python -m pip install -e . + +UIG_STATE_DIR="$PWD/.intentgate-state" .venv/bin/uig -- git status +``` + +Allowed commands stay quiet by default. Use `--explain` or `--json` to inspect a decision, and `--dry-run` to assess without executing. + +```text +intentgate: BLOCK risk=100 + +90 destructive-action: Matched destructive catalog action(s): shadow-copy-delete. + +25 admin-risk-amplifier: A risky command is being executed from an elevated Windows administrator session. + +10 missing-purpose: No purpose was supplied for a risky operation. +``` + +### CLI reference + +| Command | Purpose | +|---|---| +| `uig -- ` | Assess and execute a command | +| `uig --dry-run --explain -- ` | Assess without executing | +| `uig --purpose "..." -- ` | Supply the user's stated intent | +| `uig --shell -- ` | Explicitly permit shell operators such as pipes | +| `uig-scan .` | Refresh cached project provenance signals | +| `uig-service` | Start signal-ingestion and metrics endpoints | +| `uig-collector config/integrations.json` | Poll configured security sources | +| `uig-notifier` | Deliver queued manager risk reports | + +Exit codes are `0` for success, `2` for a non-allow dry run, `125` for review not approved, `126` for a blocked command, and `127` when the executable is missing. + +## Risk signals + +The current engine detects and correlates: + +- Recursive or forced deletion and broad filesystem targets +- Filesystem formatting, raw-disk writes, partition and boot changes +- Service shutdown, firewall modification, and network resets +- User, group, ownership, ACL, and permission changes +- Scheduled persistence and security-control disabling +- Destructive SQL, container pruning, cluster deletion, and infrastructure destruction +- Cloud deletion and Git history rewrites +- Windows registry, event-log, and shadow-copy removal +- Download-to-execution pipelines, obfuscation, secret access, and potential exfiltration +- Publishing or deployment from dirty or apparently untested repositories +- Privileged execution, unusual command sequences, and external security posture + +The catalog lives in [`src/intentgate/catalog.py`](src/intentgate/catalog.py) and is designed to grow into shell-specific AST policies. + +## Security integrations + +All integrations normalize into a small common signal: + +```json +{ + "source": "microsoft-defender-xdr", + "event_id": "alert-123", + "score": 90, + "confidence": 0.95, + "ttl_seconds": 300, + "detail": "Credential theft behavior detected", + "scope": {"device": "EXAMPLE-HOST", "user": "example-user"} +} +``` + +| Integration | Method | Template status | +|---|---|---| +| Microsoft Defender Antivirus | Local PowerShell collector | Included | +| Microsoft Defender XDR | Authenticated REST polling | Included | +| Microsoft Sentinel | Authenticated REST polling | Included | +| CrowdStrike Falcon | Normalizer endpoint | Included | +| SentinelOne | Authenticated REST polling | Included | +| Splunk Enterprise Security | Export/search polling | Included | +| Elastic Security | Detection alert search | Included | +| Any security product | `POST /v1/signals` webhook | Supported | + +Signals are deduplicated, confidence-weighted, scoped, and expired using TTLs. Vendor network calls never occur in the execution path. See [Integration Guide](docs/INTEGRATIONS.md). + +## Grafana dashboard + +Copy the example configuration and launch the observability stack: + +```powershell +Copy-Item config\integrations.example.json config\integrations.json +Copy-Item .env.integrations.example .env.integrations +docker compose --env-file .env.integrations -f docker-compose.observability.yml up -d --build +``` + +| Service | Default URL | +|---|---| +| Grafana | `http://localhost:3000` | +| Prometheus | `http://localhost:9090` | +| Intent Gate API | `http://localhost:8787` | + +The dashboard tracks aggregate security posture, active external signals, decision counts, policy latency, source-level risk, audited commands, and pending manager reports. + +> [!WARNING] +> Development credentials are intentionally simple. Change all passwords and tokens before exposing any service beyond localhost. + +## Manager escalation + +High-risk, blocked, or strongly anomalous commands create a local risk report. Likely passwords, tokens, API keys, bearer values, URL credentials, and common secret assignments are redacted before delivery. + +```powershell +$env:UIG_MANAGER_ID = "security-operations" +$env:UIG_MANAGER_REPORT_THRESHOLD = "70" +$env:UIG_MANAGER_WEBHOOK_URL = "https://webhook.example.invalid/security" +$env:UIG_MANAGER_WEBHOOK_STYLE = "generic" # generic, slack, or teams +uig-notifier +``` + +Reporting is disabled until explicitly configured, and command execution never waits for delivery. Organizations should establish employee notice, retention, access control, and appeal procedures before deployment. + +## Infrastructure as code + +The repository includes two supported automation paths. Both reuse the canonical Compose definition instead of maintaining divergent copies of the application stack. + +### Terraform + +The [`deploy/terraform`](deploy/terraform) root module uses the `kreuzwerker/docker` provider's native `docker_compose` resource. It supports local engines, named Docker contexts, remote daemon URIs, optional environment files, Compose profiles, health waiting, and clean Terraform-managed teardown. + +```bash +cd deploy/terraform +terraform init +terraform plan +terraform apply +``` + +### Ansible + +The [`deploy/ansible`](deploy/ansible) role can bootstrap Docker Engine and Compose v2 on a Debian-family server, check out an approved version, render a protected environment file from Ansible Vault variables, deploy the stack, and verify the Intent Gate health endpoint. + +```bash +cd deploy/ansible +ansible-galaxy collection install -r requirements.yml +ansible-playbook -i inventories/production.ini deploy.yml --ask-vault-pass +``` + +Terraform state, local variable files, production inventories, host variables, Vault material, and rendered secrets are excluded from Git. See the individual Terraform and Ansible guides for remote-host and lifecycle details. + +## Development + +```powershell +python -m venv .venv +.\.venv\Scripts\python.exe -m pip install -e . +.\.venv\Scripts\python.exe -m unittest discover -s tests -v +.\.venv\Scripts\python.exe -m compileall -q src tests +docker compose -f docker-compose.observability.yml config --quiet +``` + +The current suite covers policy outcomes, destructive patterns, privilege amplification, behavioral baselines, external signal correlation, reporting redaction, notifier delivery, service ingestion, and Prometheus metrics. + +## Project status + +Version **0.3.0** is a research-quality proof of concept. Important production work remains: + +- Enforce at a non-bypassable execution boundary. +- Replace command regexes with PowerShell, POSIX shell, and command AST parsers. +- Resolve concrete filesystem, cloud, database, and infrastructure targets before execution. +- Add signed policy bundles, policy versioning, exception workflows, and tamper protection. +- Add explicit feed-health policy with fail-open, fail-closed, and review-only modes. +- Evaluate anomaly quality against representative benign and adversarial datasets. +- Add durable encrypted storage and enterprise identity for reports and policy administration. + +## Responsible use + +This project observes commands and can generate workplace security reports. Deploy it transparently, collect only what is necessary, protect the resulting data, and provide meaningful human review. Do not treat anomaly scores or AI-related signals as proof of malicious intent. + +## Contributing + +Contributions are welcome. Read [CONTRIBUTING.md](CONTRIBUTING.md), review the [Security Policy](SECURITY.md), and use the provided issue templates. By participating, you agree to the [Code of Conduct](CODE_OF_CONDUCT.md). + +## License + +Licensed under the [MIT License](LICENSE). diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000..dda49af --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,34 @@ +# Security Policy + +## Project status + +User Intent AI Security is a proof of concept. Do not rely on it as the sole control protecting production systems, privileged accounts, sensitive data, or safety-critical infrastructure. + +## Reporting a vulnerability + +Please do not open a public issue for a suspected vulnerability that could expose users or deployment details. Use GitHub's private vulnerability reporting feature for this repository when available. Include: + +- A clear description of the issue and potential impact +- Affected version, platform, and configuration +- Reproduction steps or a minimal proof of concept +- Any suggested mitigation + +You should receive an initial acknowledgment within seven days. Disclosure timing will be coordinated after the issue is understood and a mitigation is available. + +## Security assumptions + +- The command wrapper is bypassable unless integrated at a trusted execution boundary. +- Regex-based command classification cannot resolve every alias, script, encoded payload, child process, or shell grammar edge case. +- External feeds may be stale, unavailable, compromised, or incorrectly scoped. +- Behavioral anomalies and AI-related code signals are risk indicators, not proof of malicious intent. +- Audit history and manager reports may contain sensitive operational metadata even after secret redaction. + +## Deployment guidance + +- Bind ingestion and metrics services to trusted interfaces only. +- Replace every development credential before deployment. +- Use TLS, authenticated webhooks, least-privilege API credentials, and restricted storage permissions. +- Define retention and access policies for audit data and manager reports. +- Validate integration mappings against the deployed vendor version. +- Decide explicitly whether stale feeds fail open, fail closed, or require review. + diff --git a/config/integrations.example.json b/config/integrations.example.json new file mode 100644 index 0000000..15a3c10 --- /dev/null +++ b/config/integrations.example.json @@ -0,0 +1,90 @@ +{ + "integrations": [ + { + "name": "windows-defender", + "kind": "windows_defender", + "enabled": true, + "ttl_seconds": 120, + "threat_ttl_seconds": 3600 + }, + { + "name": "microsoft-defender-xdr", + "enabled": false, + "url": "https://api.security.microsoft.com/api/alerts?$top=100", + "bearer_token_env": "DEFENDER_XDR_TOKEN", + "records_path": "value", + "id_path": "id", + "severity_path": "severity", + "detail_path": "title", + "scope_paths": {"device": "computerDnsName"}, + "exclude_values": {"status": ["Resolved"]}, + "ttl_seconds": 300 + }, + { + "name": "microsoft-sentinel", + "enabled": false, + "url": "https://management.azure.com/subscriptions/SUB/resourceGroups/RG/providers/Microsoft.OperationalInsights/workspaces/WORKSPACE/providers/Microsoft.SecurityInsights/incidents?api-version=2024-09-01", + "bearer_token_env": "MICROSOFT_SENTINEL_TOKEN", + "records_path": "value", + "id_path": "name", + "severity_path": "properties.severity", + "detail_path": "properties.title", + "exclude_values": {"properties.status": ["Closed"]}, + "ttl_seconds": 300 + }, + { + "name": "crowdstrike-falcon", + "enabled": false, + "url": "https://YOUR-SECURITY-NORMALIZER/crowdstrike/detections", + "bearer_token_env": "CROWDSTRIKE_ACCESS_TOKEN", + "records_path": "detections", + "id_path": "id", + "severity_path": "max_severity_displayname", + "detail_path": "description", + "ttl_seconds": 300 + }, + { + "name": "sentinelone", + "enabled": false, + "url": "https://YOUR-S1-CONSOLE/web/api/v2.1/threats?limit=100", + "api_token_env": "SENTINELONE_API_TOKEN", + "api_token_header": "Authorization", + "api_token_prefix": "ApiToken ", + "records_path": "data", + "id_path": "id", + "severity_path": "threatInfo.confidenceLevel", + "detail_path": "threatInfo.threatName", + "scope_paths": {"device": "agentRealtimeInfo.agentComputerName"}, + "ttl_seconds": 300 + }, + { + "name": "splunk-enterprise-security", + "enabled": false, + "url": "https://YOUR-SPLUNK:8089/services/search/jobs/export?output_mode=json&search=search%20index%3Dnotable", + "api_token_env": "SPLUNK_TOKEN", + "api_token_header": "Authorization", + "api_token_prefix": "Splunk ", + "id_path": "event_id", + "severity_path": "severity", + "detail_path": "rule_name", + "scope_paths": {"device": "dest", "user": "user"}, + "ttl_seconds": 300 + }, + { + "name": "elastic-security", + "enabled": false, + "url": "https://YOUR-ELASTIC/api/detection_engine/signals/search", + "method": "POST", + "body": {"query": {"bool": {"filter": [{"range": {"@timestamp": {"gte": "now-5m"}}}]}}, "size": 100}, + "api_token_env": "ELASTIC_API_KEY", + "api_token_header": "Authorization", + "api_token_prefix": "ApiKey ", + "records_path": "hits.hits", + "id_path": "_id", + "severity_path": "_source.kibana.alert.severity", + "detail_path": "_source.kibana.alert.rule.name", + "scope_paths": {"device": "_source.host.name", "user": "_source.user.name"}, + "ttl_seconds": 300 + } + ] +} diff --git a/deploy/ansible/README.md b/deploy/ansible/README.md new file mode 100644 index 0000000..41b1781 --- /dev/null +++ b/deploy/ansible/README.md @@ -0,0 +1,35 @@ +# Ansible deployment + +This playbook configures a Debian-family Linux host, optionally installs Docker Engine with Compose v2, checks out an approved repository version, renders protected environment configuration, deploys the stack, and verifies its health endpoint. + +## Controller requirements + +- Ansible Core 2.15+ +- `community.docker` collection 5.2+ +- SSH access and privilege escalation on the target + +```bash +cd deploy/ansible +ansible-galaxy collection install -r requirements.yml +``` + +## Configure + +1. Copy `inventories/example.ini` to an untracked inventory. +2. Copy `group_vars/all.example.yml` to `group_vars/all.yml`. +3. Replace example values and encrypt the variable file: + +```bash +ansible-vault encrypt group_vars/all.yml +``` + +Pin `intentgate_version` to an approved tag or commit for production deployments. Set `intentgate_manage_docker: false` on non-Debian hosts where Docker Engine and Compose v2 are already managed separately. + +## Deploy + +```bash +ansible-playbook -i inventories/production.ini deploy.yml --ask-vault-pass +``` + +The role avoids printing deployment secrets with `no_log`. Review your Ansible callback, fact cache, and controller logging configuration before using real credentials. + diff --git a/deploy/ansible/ansible.cfg b/deploy/ansible/ansible.cfg new file mode 100644 index 0000000..326aaf0 --- /dev/null +++ b/deploy/ansible/ansible.cfg @@ -0,0 +1,10 @@ +[defaults] +inventory = inventories/example.ini +roles_path = roles +host_key_checking = True +retry_files_enabled = False +interpreter_python = auto_silent + +[ssh_connection] +pipelining = True + diff --git a/deploy/ansible/deploy.yml b/deploy/ansible/deploy.yml new file mode 100644 index 0000000..c8b9a27 --- /dev/null +++ b/deploy/ansible/deploy.yml @@ -0,0 +1,8 @@ +--- +- name: Configure and deploy User Intent AI Security + hosts: intentgate_hosts + become: true + gather_facts: true + + roles: + - role: intentgate diff --git a/deploy/ansible/group_vars/all.example.yml b/deploy/ansible/group_vars/all.example.yml new file mode 100644 index 0000000..2157732 --- /dev/null +++ b/deploy/ansible/group_vars/all.example.yml @@ -0,0 +1,17 @@ +--- +# Copy to group_vars/all.yml, encrypt it with Ansible Vault, and never commit it. +intentgate_version: "main" +intentgate_install_dir: "/opt/user-intent-ai-security" +intentgate_manage_docker: true +intentgate_compose_profiles: [] + +intentgate_ingest_token: "replace-with-vault-secret" +intentgate_grafana_admin_user: "admin" +intentgate_grafana_admin_password: "replace-with-vault-secret" + +intentgate_manager_id: "security-operations" +intentgate_manager_report_threshold: 70 +intentgate_manager_webhook_url: "" +intentgate_manager_webhook_token: "" +intentgate_manager_webhook_style: "generic" + diff --git a/deploy/ansible/inventories/example.ini b/deploy/ansible/inventories/example.ini new file mode 100644 index 0000000..34c0598 --- /dev/null +++ b/deploy/ansible/inventories/example.ini @@ -0,0 +1,3 @@ +[intentgate_hosts] +example-host ansible_host=host.example.invalid ansible_user=automation + diff --git a/deploy/ansible/requirements.yml b/deploy/ansible/requirements.yml new file mode 100644 index 0000000..233e57d --- /dev/null +++ b/deploy/ansible/requirements.yml @@ -0,0 +1,4 @@ +collections: + - name: community.docker + version: ">=5.2.1" + diff --git a/deploy/ansible/roles/intentgate/defaults/main.yml b/deploy/ansible/roles/intentgate/defaults/main.yml new file mode 100644 index 0000000..f690d2e --- /dev/null +++ b/deploy/ansible/roles/intentgate/defaults/main.yml @@ -0,0 +1,20 @@ +--- +intentgate_repository_url: "https://github.com/BB-AI-Arena/User-Intent-AI-Security.git" +intentgate_version: "main" +intentgate_install_dir: "/opt/user-intent-ai-security" +intentgate_manage_docker: true +intentgate_compose_profiles: [] +intentgate_pull_repository: true +intentgate_docker_architecture_map: + x86_64: amd64 + aarch64: arm64 + armv7l: armhf + +intentgate_ingest_token: "" +intentgate_manager_id: "" +intentgate_manager_report_threshold: 70 +intentgate_manager_webhook_url: "" +intentgate_manager_webhook_token: "" +intentgate_manager_webhook_style: "generic" +intentgate_grafana_admin_user: "admin" +intentgate_grafana_admin_password: "" diff --git a/deploy/ansible/roles/intentgate/handlers/main.yml b/deploy/ansible/roles/intentgate/handlers/main.yml new file mode 100644 index 0000000..bb27e7c --- /dev/null +++ b/deploy/ansible/roles/intentgate/handlers/main.yml @@ -0,0 +1,11 @@ +--- +- name: Restart Intent Gate stack + community.docker.docker_compose_v2: + project_src: "{{ intentgate_install_dir }}" + files: + - docker-compose.observability.yml + env_files: + - .env.integrations + profiles: "{{ intentgate_compose_profiles }}" + state: restarted + no_log: true diff --git a/deploy/ansible/roles/intentgate/tasks/main.yml b/deploy/ansible/roles/intentgate/tasks/main.yml new file mode 100644 index 0000000..ffb4781 --- /dev/null +++ b/deploy/ansible/roles/intentgate/tasks/main.yml @@ -0,0 +1,149 @@ +--- +- name: Validate required deployment secrets + ansible.builtin.assert: + that: + - intentgate_ingest_token | length >= 24 + - intentgate_grafana_admin_password | length >= 12 + fail_msg: >- + Set a unique ingest token of at least 24 characters and a Grafana administrator + password of at least 12 characters, preferably through Ansible Vault. + no_log: true + +- name: Validate supported automatic Docker installation platform + ansible.builtin.assert: + that: + - ansible_os_family == "Debian" + fail_msg: >- + Automatic Docker installation currently supports Debian-family hosts. + Set intentgate_manage_docker=false when Docker Engine and Compose v2 are already installed. + when: intentgate_manage_docker | bool + +- name: Install base operating-system packages + ansible.builtin.apt: + name: + - ca-certificates + - curl + - git + - python3-debian + state: present + update_cache: true + when: intentgate_manage_docker | bool + +- name: Create Docker APT keyring directory + ansible.builtin.file: + path: /etc/apt/keyrings + state: directory + owner: root + group: root + mode: "0755" + when: intentgate_manage_docker | bool + +- name: Install Docker repository signing key + ansible.builtin.get_url: + url: "https://download.docker.com/linux/{{ ansible_distribution | lower }}/gpg" + dest: /etc/apt/keyrings/docker.asc + owner: root + group: root + mode: "0644" + when: intentgate_manage_docker | bool + +- name: Configure the Docker package repository + ansible.builtin.deb822_repository: + name: docker + types: [deb] + uris: "https://download.docker.com/linux/{{ ansible_distribution | lower }}" + suites: ["{{ ansible_distribution_release }}"] + components: [stable] + architectures: ["{{ intentgate_docker_architecture_map.get(ansible_architecture, ansible_architecture) }}"] + signed_by: /etc/apt/keyrings/docker.asc + state: present + register: intentgate_docker_repository + when: intentgate_manage_docker | bool + +- name: Install Docker Engine and Compose v2 + ansible.builtin.apt: + name: + - containerd.io + - docker-buildx-plugin + - docker-ce + - docker-ce-cli + - docker-compose-plugin + state: present + update_cache: "{{ intentgate_docker_repository.changed | default(false) }}" + when: intentgate_manage_docker | bool + +- name: Enable and start Docker Engine + ansible.builtin.service: + name: docker + enabled: true + state: started + when: intentgate_manage_docker | bool + +- name: Check out the approved Intent Gate version + ansible.builtin.git: + repo: "{{ intentgate_repository_url }}" + dest: "{{ intentgate_install_dir }}" + version: "{{ intentgate_version }}" + update: "{{ intentgate_pull_repository }}" + force: false + notify: Restart Intent Gate stack + +- name: Create private runtime state directory + ansible.builtin.file: + path: "{{ intentgate_install_dir }}/.intentgate-state" + state: directory + owner: root + group: root + mode: "0700" + +- name: Check for integration configuration + ansible.builtin.stat: + path: "{{ intentgate_install_dir }}/config/integrations.json" + register: intentgate_integrations_file + +- name: Seed disabled integration configuration + ansible.builtin.copy: + src: "{{ intentgate_install_dir }}/config/integrations.example.json" + dest: "{{ intentgate_install_dir }}/config/integrations.json" + remote_src: true + owner: root + group: root + mode: "0600" + when: not intentgate_integrations_file.stat.exists + +- name: Render protected deployment environment + ansible.builtin.template: + src: env.integrations.j2 + dest: "{{ intentgate_install_dir }}/.env.integrations" + owner: root + group: root + mode: "0600" + no_log: true + notify: Restart Intent Gate stack + +- name: Deploy Intent Gate with Docker Compose v2 + community.docker.docker_compose_v2: + project_src: "{{ intentgate_install_dir }}" + files: + - docker-compose.observability.yml + env_files: + - .env.integrations + profiles: "{{ intentgate_compose_profiles }}" + state: present + build: always + pull: policy + remove_orphans: true + wait: true + wait_timeout: 120 + no_log: true + +- name: Verify the Intent Gate health endpoint + ansible.builtin.uri: + url: http://127.0.0.1:8787/healthz + method: GET + status_code: 200 + return_content: true + register: intentgate_health + retries: 10 + delay: 3 + until: intentgate_health.json.status | default('') == 'ok' diff --git a/deploy/ansible/roles/intentgate/templates/env.integrations.j2 b/deploy/ansible/roles/intentgate/templates/env.integrations.j2 new file mode 100644 index 0000000..9ce0895 --- /dev/null +++ b/deploy/ansible/roles/intentgate/templates/env.integrations.j2 @@ -0,0 +1,9 @@ +UIG_INGEST_TOKEN={{ intentgate_ingest_token | quote }} +UIG_MANAGER_ID={{ intentgate_manager_id | quote }} +UIG_MANAGER_REPORT_THRESHOLD={{ intentgate_manager_report_threshold }} +UIG_MANAGER_WEBHOOK_URL={{ intentgate_manager_webhook_url | quote }} +UIG_MANAGER_WEBHOOK_TOKEN={{ intentgate_manager_webhook_token | quote }} +UIG_MANAGER_WEBHOOK_STYLE={{ intentgate_manager_webhook_style | quote }} +GRAFANA_ADMIN_USER={{ intentgate_grafana_admin_user | quote }} +GRAFANA_ADMIN_PASSWORD={{ intentgate_grafana_admin_password | quote }} + diff --git a/deploy/terraform/.terraform.lock.hcl b/deploy/terraform/.terraform.lock.hcl new file mode 100644 index 0000000..5660ed1 --- /dev/null +++ b/deploy/terraform/.terraform.lock.hcl @@ -0,0 +1,22 @@ +# This file is maintained automatically by "terraform init". +# Manual edits may be lost in future updates. + +provider "registry.terraform.io/kreuzwerker/docker" { + version = "4.5.0" + constraints = "~> 4.5" + hashes = [ + "h1:mEARBnOnl9C1tcmrdNS+79cDiSUVBFX4gnwGEm4UQcE=", + "zh:0ee4e9121632158a9ae1049bdd53c71fd9e06357de915dd193df2c1c55ae8814", + "zh:12c6991012b5a548a406e1960c323484f95a9538c2522af15ec82f47fdb832d1", + "zh:19c86ae59ed1e062e8402a49ea842f957fcda2f9689e06ecbd9480cc9bd1e41a", + "zh:3ac8bc9805acd20c40c0ed474c9f4605aa73b61cf52809e1131e335fcbfe8e49", + "zh:5a0e70d143759712fdda109f024a41d59d4758b5615378aa21ee6b959f65a416", + "zh:5d3599285f71cc53c88a5802fa6e2f8ba87b9deea03abe5fe8bb2639e9d95654", + "zh:a42f573a2526cddc9b8a93637a2c5478bc6e903ff29556ab8b2f1467681d5bde", + "zh:c8de50d902cc8457e565a5d959f1e3e49064d31c2f14f7adf5f3ec632cfe3695", + "zh:d40543d51eb4e902125d2c5a8a0f425a60edcec6c1dc07c4a89d92335ec05541", + "zh:dd3105018ffddc8114083dcf84798c32fdecca7ef708eceaee49eb4b306fc061", + "zh:e51e0199700698b962085683af4551d1f982dce37f3b6950da062c65e0145505", + "zh:ffe415d3d3deffffdbd3a7173e3a9f90b74601cb6d8acafec1ab701a5b91dbff", + ] +} diff --git a/deploy/terraform/README.md b/deploy/terraform/README.md new file mode 100644 index 0000000..66f409b --- /dev/null +++ b/deploy/terraform/README.md @@ -0,0 +1,32 @@ +# Terraform deployment + +This root module manages the existing Docker Compose application through the `kreuzwerker/docker` provider. It intentionally reuses `docker-compose.observability.yml` so service definitions remain canonical in one place. + +## Requirements + +- Terraform 1.6+ +- A reachable Docker Engine +- Docker provider 4.5+ + +## Deploy + +```bash +cd deploy/terraform +terraform init +terraform plan +terraform apply +``` + +For an alternate engine, copy `terraform.tfvars.example` to an untracked `terraform.tfvars` and set either `docker_context` or `docker_host`. + +To enable collectors or notifications, first create the untracked environment and integration configuration described in the main README, then set: + +```hcl +environment_file = "../../.env.integrations" +profiles = ["collectors", "notifications"] +``` + +Terraform state can contain infrastructure metadata and must be stored in an approved encrypted backend for shared or production use. State files and local variable files are excluded from Git. + +> Do not use Terraform to adopt the same Compose project while it is already managed independently. Stop the manually managed stack first or select a different `project_name`. + diff --git a/deploy/terraform/main.tf b/deploy/terraform/main.tf new file mode 100644 index 0000000..8b46dc0 --- /dev/null +++ b/deploy/terraform/main.tf @@ -0,0 +1,17 @@ +locals { + repository_root = abspath("${path.module}/../..") + compose_file = "${local.repository_root}/docker-compose.observability.yml" + env_files = var.environment_file == null ? [] : [abspath(var.environment_file)] +} + +resource "docker_compose" "intentgate" { + project_name = var.project_name + project_directory = local.repository_root + config_paths = [local.compose_file] + env_files = local.env_files + profiles = var.profiles + remove_orphans = true + wait = true + wait_timeout = var.wait_timeout +} + diff --git a/deploy/terraform/outputs.tf b/deploy/terraform/outputs.tf new file mode 100644 index 0000000..1d582e9 --- /dev/null +++ b/deploy/terraform/outputs.tf @@ -0,0 +1,20 @@ +output "grafana_url" { + description = "Default local Grafana URL." + value = "http://localhost:3000" +} + +output "prometheus_url" { + description = "Default local Prometheus URL." + value = "http://localhost:9090" +} + +output "intentgate_api_url" { + description = "Default local Intent Gate API URL." + value = "http://localhost:8787" +} + +output "compose_project" { + description = "Managed Docker Compose project identifier." + value = docker_compose.intentgate.id +} + diff --git a/deploy/terraform/providers.tf b/deploy/terraform/providers.tf new file mode 100644 index 0000000..5a5f9fa --- /dev/null +++ b/deploy/terraform/providers.tf @@ -0,0 +1,5 @@ +provider "docker" { + host = var.docker_host + context = var.docker_context +} + diff --git a/deploy/terraform/terraform.tfvars.example b/deploy/terraform/terraform.tfvars.example new file mode 100644 index 0000000..6c6a714 --- /dev/null +++ b/deploy/terraform/terraform.tfvars.example @@ -0,0 +1,17 @@ +project_name = "intentgate-example" + +# Use a named Docker context for a remote or alternate engine. +# docker_context = "example-context" + +# Or provide a daemon URI. Do not disable SSH host-key checking. +# docker_host = "ssh://automation@example-host:22" + +# Keep this file outside version control when it contains deployment values. +# environment_file = "../../.env.integrations" + +# Enable only after the corresponding configuration and secrets exist. +profiles = [] +# profiles = ["collectors", "notifications"] + +wait_timeout = "2m" + diff --git a/deploy/terraform/variables.tf b/deploy/terraform/variables.tf new file mode 100644 index 0000000..c051063 --- /dev/null +++ b/deploy/terraform/variables.tf @@ -0,0 +1,49 @@ +variable "project_name" { + description = "Docker Compose project name used to namespace containers, networks, and volumes." + type = string + default = "user-intent-ai-security" + + validation { + condition = can(regex("^[a-z0-9][a-z0-9_-]*$", var.project_name)) + error_message = "project_name must contain only lowercase letters, numbers, underscores, and hyphens." + } +} + +variable "docker_host" { + description = "Optional Docker daemon URI. Leave null to use the provider or DOCKER_HOST default." + type = string + default = null + nullable = true +} + +variable "docker_context" { + description = "Optional Docker CLI context name. When set, it takes precedence over docker_host." + type = string + default = null + nullable = true +} + +variable "environment_file" { + description = "Optional path to a local, untracked Compose environment file containing deployment secrets." + type = string + default = null + nullable = true +} + +variable "profiles" { + description = "Optional Compose profiles to enable, such as collectors and notifications." + type = list(string) + default = [] +} + +variable "wait_timeout" { + description = "Maximum time for the Docker provider to wait for services during apply." + type = string + default = "2m" + + validation { + condition = can(regex("^[1-9][0-9]*(s|m|h)$", var.wait_timeout)) + error_message = "wait_timeout must be a positive duration such as 30s, 2m, or 1h." + } +} + diff --git a/deploy/terraform/versions.tf b/deploy/terraform/versions.tf new file mode 100644 index 0000000..f5ff514 --- /dev/null +++ b/deploy/terraform/versions.tf @@ -0,0 +1,11 @@ +terraform { + required_version = ">= 1.6.0" + + required_providers { + docker = { + source = "kreuzwerker/docker" + version = "~> 4.5" + } + } +} + diff --git a/docker-compose.observability.yml b/docker-compose.observability.yml new file mode 100644 index 0000000..a581bac --- /dev/null +++ b/docker-compose.observability.yml @@ -0,0 +1,68 @@ +services: + intentgate: + build: . + environment: + UIG_STATE_DIR: /state + UIG_INGEST_TOKEN: ${UIG_INGEST_TOKEN:-local-dev-change-me} + volumes: + - ./.intentgate-state:/state + ports: + - "8787:8787" + restart: unless-stopped + + prometheus: + image: prom/prometheus:latest + command: ["--config.file=/etc/prometheus/prometheus.yml"] + volumes: + - ./observability/prometheus.yml:/etc/prometheus/prometheus.yml:ro + - prometheus-data:/prometheus + ports: + - "9090:9090" + depends_on: [intentgate] + restart: unless-stopped + + grafana: + image: grafana/grafana:latest + environment: + GF_SECURITY_ADMIN_USER: ${GRAFANA_ADMIN_USER:-admin} + GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_ADMIN_PASSWORD:-intentgate} + GF_USERS_ALLOW_SIGN_UP: "false" + volumes: + - ./observability/grafana/provisioning:/etc/grafana/provisioning:ro + - ./observability/grafana/dashboards:/var/lib/grafana/dashboards:ro + - grafana-data:/var/lib/grafana + ports: + - "3000:3000" + depends_on: [prometheus] + restart: unless-stopped + + collector: + build: . + command: ["uig-collector", "/config/integrations.json", "--interval", "30"] + profiles: ["collectors"] + environment: + UIG_STATE_DIR: /state + env_file: + - path: .env.integrations + required: false + volumes: + - ./.intentgate-state:/state + - ./config/integrations.json:/config/integrations.json:ro + restart: unless-stopped + + notifier: + build: . + command: ["uig-notifier", "--interval", "10"] + profiles: ["notifications"] + environment: + UIG_STATE_DIR: /state + UIG_MANAGER_WEBHOOK_URL: ${UIG_MANAGER_WEBHOOK_URL:-} + UIG_MANAGER_WEBHOOK_TOKEN: ${UIG_MANAGER_WEBHOOK_TOKEN:-} + UIG_MANAGER_WEBHOOK_STYLE: ${UIG_MANAGER_WEBHOOK_STYLE:-generic} + volumes: + - ./.intentgate-state:/state + restart: unless-stopped + +volumes: + prometheus-data: + grafana-data: diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md new file mode 100644 index 0000000..6e6bac0 --- /dev/null +++ b/docs/ARCHITECTURE.md @@ -0,0 +1,35 @@ +# Architecture + +## Design goal + +The command path should remain fast enough that users do not notice the gate during ordinary work. The architecture separates synchronous policy evaluation from every operation that can block on a model, vendor API, network, or notification service. + +## Components + +| Component | Responsibility | Execution path | +|---|---|---| +| `context.py` | Collect bounded local user, Git, project, and cached posture context | Synchronous | +| `catalog.py` / `rules.py` | Classify behavior and calculate risk signals | Synchronous | +| `engine.py` | Aggregate signals into `ALLOW`, `REVIEW`, or `BLOCK` | Synchronous | +| `behavior.py` | Compare a command family with recent per-user history | Synchronous local read | +| `provenance.py` | Scan source health and provenance indicators | Asynchronous/manual | +| `collector.py` | Poll AV, EDR, and SIEM APIs | Background | +| `integrations.py` | Normalize, scope, deduplicate, expire, and correlate signals | Background writes; cached reads | +| `reporting.py` | Redact and queue high-risk reports | Local append after decision | +| `notifier.py` | Deliver reports to approved webhooks | Background | +| `service.py` | Ingest signals and expose posture and Prometheus metrics | Background | + +## Data flow + +External signals receive a score, confidence, scope, and TTL. The correlation layer takes the highest effective signal and adds bounded corroboration from other sources. The gate reads only non-expired local state. + +Audit and report files are local JSON/JSONL in `UIG_STATE_DIR`. The PowerShell helper defaults this to `.intentgate-state` in the project so the host CLI and Docker observability services share the same state. + +## Latency model + +No external API is called during a command decision. The development benchmark for context collection and scoring is approximately 0.53 ms in-process. Python process startup is separate; production deployments should use a persistent broker or daemon. + +## Enforcement boundary + +The CLI wrapper demonstrates policy behavior but is bypassable. Production integrations should place the engine inside the agent tool broker, shell host, endpoint service, or privileged API that owns process creation. + diff --git a/docs/INTEGRATIONS.md b/docs/INTEGRATIONS.md new file mode 100644 index 0000000..8d8a4de --- /dev/null +++ b/docs/INTEGRATIONS.md @@ -0,0 +1,48 @@ +# Security Integration Guide + +## Normalized signal contract + +Send a JSON object or array to `POST /v1/signals`: + +```json +{ + "source": "security-product", + "event_id": "stable-alert-id", + "score": 85, + "confidence": 0.9, + "ttl_seconds": 300, + "detail": "Short operator-readable explanation", + "scope": { + "device": "HOST-01", + "user": "example-user", + "project": "/example/project" + } +} +``` + +`score` is clamped to 0–100, `confidence` to 0–1, and TTL to one second through seven days. Scope keys are optional. Events are deduplicated by source and event ID. + +## Authentication + +Set `UIG_INGEST_TOKEN` and include: + +```text +Authorization: Bearer +``` + +Without a configured token, ingestion is accepted only from loopback addresses. Production deployments should use TLS and network-level access controls as well. + +## Collector configuration + +Copy `config/integrations.example.json` to `config/integrations.json`, enable only required sources, and provide credentials through environment variables. Never store API credentials in the JSON file. + +```powershell +uig-collector config\integrations.json --once +uig-collector config\integrations.json --interval 30 +``` + +Templates are starting points. Verify endpoints, authorization flows, response fields, severity mappings, filters, pagination, and rate limits against the exact tenant and product version before enforcement. + +## Correlation + +The highest confidence-weighted active signal forms the base posture. Additional independent signals add bounded corroboration. Expired signals stop contributing. The POC currently fails open when feeds are absent or stale; production policy should make feed health explicit. diff --git a/docs/THREAT_MODEL.md b/docs/THREAT_MODEL.md new file mode 100644 index 0000000..81827bf --- /dev/null +++ b/docs/THREAT_MODEL.md @@ -0,0 +1,42 @@ +# Threat Model + +## Protected outcomes + +- Prevent unintended destructive or externally consequential commands. +- Increase scrutiny when elevated privilege, unusual behavior, or external security alerts coincide. +- Preserve sufficient redacted evidence for human review. +- Keep ordinary command latency low and vendor failures isolated. + +## In-scope threats + +- Accidental destructive commands and excessive blast radius +- Malicious or compromised user sessions +- Agent-generated commands that diverge from the stated task +- Download-and-execute pipelines and encoded execution +- Secret access followed by potential exfiltration +- Security-control weakening, log erasure, persistence, and recovery deletion +- Risky commands during an active AV, EDR, or SIEM incident + +## Out of scope for the POC + +- Kernel-level process prevention +- Complete shell parsing and script interpretation +- Memory-only attacks and kernel exploits +- Trustworthy attribution of whether code was AI-generated +- Protection when users bypass the wrapper +- Guaranteed correctness of third-party security feeds + +## Trust boundaries + +The local state directory, policy code, process-launch integration, manager webhook, and integration credentials are trusted. Production deployments must protect them against unauthorized modification. + +## Known failure modes + +- False positives from legitimate but rare administrative work +- False negatives from aliases, custom scripts, indirect execution, or novel syntax +- Stale or mis-scoped external signals +- Sensitive operational text surviving imperfect redaction +- Baseline poisoning by repeated malicious behavior + +Mitigations include human review, signed policies, protected baselines, explicit feed-health rules, shell-specific AST parsing, target resolution, and retention controls. + diff --git a/docs/assets/hero.svg b/docs/assets/hero.svg new file mode 100644 index 0000000..07e837d --- /dev/null +++ b/docs/assets/hero.svg @@ -0,0 +1,28 @@ + + User Intent AI Security + A command prompt protected by a glowing shield and contextual security signals. + + + + + + + + + + + + + + + + + + + + + $ intentgate assess + USER INTENT AI SECURITY + Context-aware protection before commands execute. + + diff --git a/observability/grafana/dashboards/intentgate.json b/observability/grafana/dashboards/intentgate.json new file mode 100644 index 0000000..0b3bd54 --- /dev/null +++ b/observability/grafana/dashboards/intentgate.json @@ -0,0 +1,87 @@ +{ + "annotations": {"list": []}, + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 1, + "id": null, + "links": [], + "panels": [ + { + "datasource": {"type": "prometheus", "uid": "prometheus"}, + "fieldConfig": {"defaults": {"max": 100, "min": 0, "thresholds": {"mode": "absolute", "steps": [{"color": "green", "value": null}, {"color": "yellow", "value": 40}, {"color": "orange", "value": 70}, {"color": "red", "value": 90}]}}}, + "gridPos": {"h": 8, "w": 6, "x": 0, "y": 0}, + "id": 1, + "options": {"reduceOptions": {"calcs": ["lastNotNull"], "values": false}, "showThresholdLabels": false, "showThresholdMarkers": true}, + "targets": [{"expr": "intentgate_security_posture", "refId": "A"}], + "title": "External Security Posture", + "type": "gauge" + }, + { + "datasource": {"type": "prometheus", "uid": "prometheus"}, + "fieldConfig": {"defaults": {"unit": "ms"}}, + "gridPos": {"h": 8, "w": 6, "x": 6, "y": 0}, + "id": 2, + "options": {"reduceOptions": {"calcs": ["lastNotNull"], "values": false}}, + "targets": [{"expr": "intentgate_decision_latency_ms", "refId": "A"}], + "title": "Policy Engine Latency", + "type": "stat" + }, + { + "datasource": {"type": "prometheus", "uid": "prometheus"}, + "fieldConfig": {"defaults": {}}, + "gridPos": {"h": 8, "w": 6, "x": 12, "y": 0}, + "id": 3, + "options": {"reduceOptions": {"calcs": ["lastNotNull"], "values": false}}, + "targets": [{"expr": "intentgate_active_external_signals", "refId": "A"}], + "title": "Active AV / SIEM Signals", + "type": "stat" + }, + { + "datasource": {"type": "prometheus", "uid": "prometheus"}, + "fieldConfig": {"defaults": {}}, + "gridPos": {"h": 8, "w": 6, "x": 18, "y": 0}, + "id": 4, + "options": {"reduceOptions": {"calcs": ["lastNotNull"], "values": false}}, + "targets": [{"expr": "intentgate_audit_events", "refId": "A"}], + "title": "Audited Commands", + "type": "stat" + }, + { + "datasource": {"type": "prometheus", "uid": "prometheus"}, + "fieldConfig": {"defaults": {"max": 100, "min": 0}}, + "gridPos": {"h": 9, "w": 12, "x": 0, "y": 8}, + "id": 5, + "targets": [{"expr": "intentgate_source_risk", "legendFormat": "{{source}}", "refId": "A"}], + "title": "Risk by Integration", + "type": "timeseries" + }, + { + "datasource": {"type": "prometheus", "uid": "prometheus"}, + "fieldConfig": {"defaults": {}}, + "gridPos": {"h": 9, "w": 12, "x": 12, "y": 8}, + "id": 6, + "targets": [{"expr": "intentgate_decisions_total", "legendFormat": "{{decision}}", "refId": "A"}], + "title": "Gate Decisions", + "type": "timeseries" + }, + { + "datasource": {"type": "prometheus", "uid": "prometheus"}, + "fieldConfig": {"defaults": {"thresholds": {"mode": "absolute", "steps": [{"color": "green", "value": null}, {"color": "orange", "value": 1}, {"color": "red", "value": 10}]}}}, + "gridPos": {"h": 6, "w": 6, "x": 0, "y": 17}, + "id": 7, + "options": {"reduceOptions": {"calcs": ["lastNotNull"], "values": false}}, + "targets": [{"expr": "intentgate_pending_manager_reports", "refId": "A"}], + "title": "Pending Manager Reports", + "type": "stat" + } + ], + "refresh": "5s", + "schemaVersion": 40, + "tags": ["security", "intentgate"], + "templating": {"list": []}, + "time": {"from": "now-1h", "to": "now"}, + "timezone": "browser", + "title": "User Intent Gate", + "uid": "user-intent-gate", + "version": 1 +} diff --git a/observability/grafana/provisioning/dashboards/intentgate.yml b/observability/grafana/provisioning/dashboards/intentgate.yml new file mode 100644 index 0000000..84281b9 --- /dev/null +++ b/observability/grafana/provisioning/dashboards/intentgate.yml @@ -0,0 +1,12 @@ +apiVersion: 1 + +providers: + - name: Intent Gate + orgId: 1 + folder: Security + type: file + disableDeletion: true + updateIntervalSeconds: 10 + options: + path: /var/lib/grafana/dashboards + diff --git a/observability/grafana/provisioning/datasources/prometheus.yml b/observability/grafana/provisioning/datasources/prometheus.yml new file mode 100644 index 0000000..00f9915 --- /dev/null +++ b/observability/grafana/provisioning/datasources/prometheus.yml @@ -0,0 +1,10 @@ +apiVersion: 1 + +datasources: + - name: Prometheus + uid: prometheus + type: prometheus + access: proxy + url: http://prometheus:9090 + isDefault: true + editable: false diff --git a/observability/prometheus.yml b/observability/prometheus.yml new file mode 100644 index 0000000..f4a5da4 --- /dev/null +++ b/observability/prometheus.yml @@ -0,0 +1,8 @@ +global: + scrape_interval: 5s + +scrape_configs: + - job_name: intentgate + static_configs: + - targets: ["intentgate:8787"] + diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d71f84c --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,23 @@ +[build-system] +requires = ["setuptools>=75"] +build-backend = "setuptools.build_meta" + +[project] +name = "user-intent-gate" +version = "0.3.0" +description = "A low-latency, context-aware command intent gate proof of concept" +readme = "README.md" +requires-python = ">=3.11" +license = { text = "MIT" } +authors = [{ name = "BB AI Arena" }] +dependencies = [] + +[project.scripts] +uig = "intentgate.cli:main" +uig-scan = "intentgate.provenance:main" +uig-service = "intentgate.service:main" +uig-collector = "intentgate.collector:main" +uig-notifier = "intentgate.notifier:main" + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/scripts/Enable-IntentGate.ps1 b/scripts/Enable-IntentGate.ps1 new file mode 100644 index 0000000..3100e4b --- /dev/null +++ b/scripts/Enable-IntentGate.ps1 @@ -0,0 +1,18 @@ +param( + [string]$ProjectRoot = (Split-Path -Parent $PSScriptRoot) +) + +$pythonPath = Join-Path $ProjectRoot ".venv\Scripts\python.exe" +if (-not (Test-Path -LiteralPath $pythonPath)) { + throw "Virtual environment not found. Run: python -m venv .venv" +} + +if (-not $env:UIG_STATE_DIR) { + $env:UIG_STATE_DIR = Join-Path $ProjectRoot ".intentgate-state" +} + +function global:uig { + & $pythonPath -m intentgate.cli @args +} + +Write-Host "Intent Gate enabled for this PowerShell session. State: $env:UIG_STATE_DIR" diff --git a/src/intentgate/__init__.py b/src/intentgate/__init__.py new file mode 100644 index 0000000..bc189aa --- /dev/null +++ b/src/intentgate/__init__.py @@ -0,0 +1,3 @@ +"""User Intent Gate proof of concept.""" + +__version__ = "0.3.0" diff --git a/src/intentgate/audit.py b/src/intentgate/audit.py new file mode 100644 index 0000000..f376e2e --- /dev/null +++ b/src/intentgate/audit.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import json +import time +from pathlib import Path + +from .context import HISTORY_FILE +from .models import Assessment, CommandContext +from .reporting import queue_manager_report + + +def record(ctx: CommandContext, result: Assessment, executed: bool, exit_code: int | None = None) -> None: + HISTORY_FILE.parent.mkdir(parents=True, exist_ok=True) + event = { + "timestamp": time.time(), + "command": ctx.command, + "cwd": ctx.cwd, + "user_name": ctx.user_name, + "privilege_level": ctx.privilege_level, + "is_root": ctx.is_root, + "is_admin": ctx.is_admin, + "anomaly_score": ctx.anomaly_score, + "purpose": ctx.purpose, + "decision": result.decision.value, + "risk_score": result.risk_score, + "latency_ms": result.latency_ms, + "signals": [signal.name for signal in result.signals], + "fingerprint": result.command_fingerprint, + "executed": executed, + "exit_code": exit_code, + } + with HISTORY_FILE.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(event, separators=(",", ":")) + "\n") + try: + queue_manager_report(ctx, result, executed, exit_code) + except OSError: + # Reporting is asynchronous and must not change the command's exit behavior. + pass diff --git a/src/intentgate/behavior.py b/src/intentgate/behavior.py new file mode 100644 index 0000000..1801b65 --- /dev/null +++ b/src/intentgate/behavior.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import json +import math +import os +import re +from collections import Counter +from pathlib import Path +from typing import Any + + +def _history_file() -> Path: + return Path(os.environ.get("UIG_STATE_DIR", Path.home() / ".intentgate")) / "history.jsonl" + + +def command_family(command: str) -> str: + tokens = re.findall(r"[a-z0-9_.-]+", command.lower()) + return " ".join(tokens[:2]) if tokens else "unknown" + + +def assess_anomaly(command: str, *, user_name: str | None = None, limit: int = 500) -> dict[str, Any]: + try: + lines = _history_file().read_text(encoding="utf-8").splitlines()[-limit:] + except OSError: + return {"score": 0, "details": [], "sample_size": 0} + families: Counter[str] = Counter() + for line in lines: + try: + item = json.loads(line) + if user_name and item.get("user_name") not in {None, user_name}: + continue + families[command_family(str(item.get("command", "")))] += 1 + except (json.JSONDecodeError, TypeError): + continue + sample_size = sum(families.values()) + if sample_size < 20: + return {"score": 0, "details": ["Behavioral baseline is still learning."], "sample_size": sample_size} + family = command_family(command) + frequency = families[family] + rarity = -math.log2((frequency + 1) / (sample_size + len(families) + 1)) + score = min(40, max(0, round((rarity - 2) * 10))) + details = [f"Command family '{family}' appeared {frequency} time(s) in {sample_size} prior gated commands."] + return {"score": score, "details": details, "sample_size": sample_size} diff --git a/src/intentgate/catalog.py b/src/intentgate/catalog.py new file mode 100644 index 0000000..4c70d5e --- /dev/null +++ b/src/intentgate/catalog.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass + + +@dataclass(frozen=True) +class DestructivePattern: + identifier: str + platforms: tuple[str, ...] + category: str + score: int + pattern: re.Pattern[str] + description: str + + +def _rx(value: str) -> re.Pattern[str]: + return re.compile(value, re.IGNORECASE) + + +DESTRUCTIVE_ACTIONS = ( + DestructivePattern("recursive-delete-unix", ("linux", "macos"), "filesystem", 55, _rx(r"(?:^|[;&|]\s*)rm\s+[^\n]*(?:-[a-z]*r[a-z]*|--recursive)\b"), "Recursively deletes files or directories."), + DestructivePattern("recursive-delete-windows", ("windows",), "filesystem", 55, _rx(r"(?:remove-item\b[^\n]*(?:-recurse|-force)|rmdir\s+/s\b|rd\s+/s\b|del\s+/[sqf])"), "Recursively or forcibly deletes Windows filesystem content."), + DestructivePattern("filesystem-format", ("all",), "storage", 90, _rx(r"(?:\bmkfs(?:\.[a-z0-9]+)?\b|\bformat(?:\.com)?\s+[a-z]:|format-volume\b)"), "Formats a filesystem or volume."), + DestructivePattern("raw-disk-write", ("all",), "storage", 90, _rx(r"(?:\bdd\b[^\n]*\bof=/dev/|clear-disk\b|initialize-disk\b|diskpart\b)"), "Writes to or reinitializes raw storage."), + DestructivePattern("partition-change", ("all",), "storage", 75, _rx(r"(?:\bfdisk\b|\bparted\b|remove-partition\b|resize-partition\b)"), "Changes disk partitions."), + DestructivePattern("boot-change", ("all",), "system", 80, _rx(r"(?:\bgrub-install\b|\bupdate-grub\b|\bbcdedit\b|\bbootrec\b)"), "Changes boot configuration."), + DestructivePattern("shutdown-reboot", ("all",), "availability", 40, _rx(r"(?:\bshutdown\b|\breboot\b|\bpoweroff\b|restart-computer\b|stop-computer\b)"), "Stops or restarts a machine."), + DestructivePattern("service-disable", ("all",), "availability", 45, _rx(r"(?:systemctl\s+(?:disable|mask|stop)\b|service\s+\S+\s+stop\b|stop-service\b|set-service\b[^\n]*disabled|sc(?:\.exe)?\s+(?:delete|stop|config)\b)"), "Stops, disables, masks, or deletes a service."), + DestructivePattern("firewall-change", ("all",), "network", 50, _rx(r"(?:iptables\b|nft\b|ufw\s+(?:disable|reset|delete)|netsh\s+advfirewall|new-netfirewallrule|remove-netfirewallrule|set-netfirewallprofile)"), "Changes host firewall policy."), + DestructivePattern("network-reset", ("all",), "network", 50, _rx(r"(?:ip\s+(?:addr|route|link)\s+(?:del|flush)|ifconfig\b[^\n]*down|netsh\s+(?:interface|winsock)\s+reset|disable-netadapter|remove-netroute)"), "Disables or resets network configuration."), + DestructivePattern("user-account-change", ("all",), "identity", 55, _rx(r"(?:userdel\b|deluser\b|usermod\b|passwd\b|net\s+user\b|remove-localuser\b|disable-localuser\b|set-localuser\b)"), "Changes or deletes a user account."), + DestructivePattern("group-membership-change", ("all",), "identity", 60, _rx(r"(?:groupdel\b|gpasswd\b|add-localgroupmember\b|remove-localgroupmember\b|net\s+localgroup\b)"), "Changes privileged or local group membership."), + DestructivePattern("permission-change", ("all",), "identity", 45, _rx(r"(?:chmod\b|chown\b|setfacl\b|icacls\b|takeown\b)"), "Changes ownership or access permissions."), + DestructivePattern("scheduled-persistence", ("all",), "persistence", 55, _rx(r"(?:crontab\b|at\b|schtasks\b|register-scheduledtask\b|new-service\b|systemctl\s+enable\b)"), "Creates or alters scheduled execution or service persistence."), + DestructivePattern("package-removal", ("all",), "software", 45, _rx(r"(?:(?:apt|apt-get|yum|dnf|pacman|zypper)\s+(?:remove|purge|autoremove)\b|uninstall-package\b|winget\s+uninstall\b|choco\s+uninstall\b)"), "Removes installed software or dependencies."), + DestructivePattern("database-destructive", ("all",), "database", 80, _rx(r"\b(?:drop\s+(?:database|schema|table)|truncate\s+table|delete\s+from\s+\S+\s*;?)\b"), "Deletes database objects or bulk data."), + DestructivePattern("container-prune", ("all",), "container", 55, _rx(r"(?:docker\s+(?:system|image|volume|container)\s+prune\b|docker\s+rm\b[^\n]*-[a-z]*f|podman\s+system\s+prune\b)"), "Removes container resources."), + DestructivePattern("cluster-delete", ("all",), "orchestration", 65, _rx(r"(?:kubectl\s+delete\b|helm\s+uninstall\b|terraform\s+destroy\b)"), "Deletes cluster or infrastructure resources."), + DestructivePattern("cloud-delete", ("all",), "cloud", 70, _rx(r"(?:(?:aws|az|gcloud)\b[^\n]*\b(?:delete|terminate|destroy|remove)\b)"), "Deletes or terminates cloud resources."), + DestructivePattern("git-history-rewrite", ("all",), "source-control", 55, _rx(r"(?:git\s+push\b[^\n]*(?:--force|-f)\b|git\s+reset\s+--hard\b|git\s+clean\b[^\n]*-[a-z]*f)"), "Rewrites remote history or discards local work."), + DestructivePattern("registry-delete", ("windows",), "system", 65, _rx(r"(?:reg(?:\.exe)?\s+delete\b|remove-item\b[^\n]*(?:hklm:|hkcu:))"), "Deletes Windows registry data."), + DestructivePattern("shadow-copy-delete", ("windows",), "recovery", 90, _rx(r"(?:vssadmin\s+delete\s+shadows|wmic\s+shadowcopy\s+delete|delete-shadowcopy)"), "Deletes recovery shadow copies."), + DestructivePattern("log-erasure", ("all",), "evasion", 75, _rx(r"(?:journalctl\s+--vacuum|wevtutil\s+cl\b|clear-eventlog\b|remove-item\b[^\n]*(?:/var/log|\\winevt\\logs))"), "Erases or truncates audit logs."), + DestructivePattern("security-control-change", ("all",), "defense-evasion", 85, _rx(r"(?:set-mppreference\b[^\n]*(?:disablerealtimemonitoring|disablebehaviormonitoring)|uninstall-windowsfeature\b[^\n]*defender|systemctl\s+(?:stop|disable|mask)\b[^\n]*(?:auditd|apparmor|selinux)|setenforce\s+0\b)"), "Disables or weakens security controls."), +) + + +def match_destructive_actions(command: str) -> list[DestructivePattern]: + return [item for item in DESTRUCTIVE_ACTIONS if item.pattern.search(command)] + diff --git a/src/intentgate/cli.py b/src/intentgate/cli.py new file mode 100644 index 0000000..3319b33 --- /dev/null +++ b/src/intentgate/cli.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import argparse +import json +import subprocess +import sys + +from .audit import record +from .context import collect_context +from .engine import assess +from .models import Decision + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="uig", description="Context-aware command intent gate") + parser.add_argument("--purpose", help="Short statement of what the user is trying to accomplish") + parser.add_argument("--explain", action="store_true", help="Print all decision signals") + parser.add_argument("--json", action="store_true", help="Print the assessment as JSON") + parser.add_argument("--dry-run", action="store_true", help="Assess without executing") + parser.add_argument("--yes", action="store_true", help="Approve REVIEW decisions non-interactively") + parser.add_argument("--shell", action="store_true", help="Execute shell operators such as pipes and redirects") + parser.add_argument("command", nargs=argparse.REMAINDER, help="Command after --") + return parser + + +def _display(result, verbose: bool) -> None: + print(f"intentgate: {result.decision.value.upper()} risk={result.risk_score} ({result.latency_ms:.3f} ms)") + if verbose: + for signal in result.signals: + sign = "+" if signal.score >= 0 else "" + print(f" {sign}{signal.score:>3} {signal.name}: {signal.detail}") + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + command = list(args.command) + if command and command[0] == "--": + command = command[1:] + if not command: + _parser().error("supply a command after --") + + ctx = collect_context(command, args.purpose) + result = assess(ctx) + if args.json: + print(json.dumps(result.to_dict(), indent=2)) + elif args.explain or result.decision is not Decision.ALLOW or args.dry_run: + _display(result, args.explain) + + if args.dry_run: + record(ctx, result, executed=False) + return 0 if result.decision is Decision.ALLOW else 2 + if result.decision is Decision.BLOCK: + record(ctx, result, executed=False) + print("intentgate: blocked; revise the command or policy", file=sys.stderr) + return 126 + if result.decision is Decision.REVIEW and not args.yes: + if not sys.stdin.isatty(): + record(ctx, result, executed=False) + print("intentgate: review required; rerun interactively or pass --yes", file=sys.stderr) + return 125 + approved = input("intentgate: allow this command? [y/N] ").strip().lower() in {"y", "yes"} + if not approved: + record(ctx, result, executed=False) + return 125 + + try: + if args.shell: + completed = subprocess.run(" ".join(command), cwd=ctx.cwd, shell=True) + else: + completed = subprocess.run(command, cwd=ctx.cwd) + code = completed.returncode + except FileNotFoundError: + record(ctx, result, executed=False) + print(f"intentgate: executable not found: {command[0]}", file=sys.stderr) + return 127 + record(ctx, result, executed=True, exit_code=code) + return code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/intentgate/collector.py b/src/intentgate/collector.py new file mode 100644 index 0000000..eafab9c --- /dev/null +++ b/src/intentgate/collector.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import time +import urllib.request +from pathlib import Path +from typing import Any + +from .integrations import ingest + + +DEFAULT_SEVERITY = { + "informational": 5, + "info": 5, + "low": 25, + "medium": 50, + "moderate": 50, + "high": 75, + "critical": 95, + "severe": 95, +} + + +def _path(value: Any, dotted: str | None, default: Any = None) -> Any: + if not dotted: + return value + current = value + for part in dotted.split("."): + if isinstance(current, dict): + current = current.get(part, default) + elif isinstance(current, list) and part.isdigit() and int(part) < len(current): + current = current[int(part)] + else: + return default + return current + + +def _headers(config: dict[str, Any]) -> dict[str, str]: + headers = {str(k): str(v) for k, v in config.get("headers", {}).items()} + if env_name := config.get("bearer_token_env"): + token = os.environ.get(str(env_name), "") + if token: + headers["Authorization"] = f"Bearer {token}" + if env_name := config.get("api_token_env"): + token = os.environ.get(str(env_name), "") + if token: + prefix = str(config.get("api_token_prefix", "")) + headers[str(config.get("api_token_header", "Authorization"))] = f"{prefix}{token}" + return headers + + +def poll_integration(config: dict[str, Any]) -> int: + if config.get("kind") == "windows_defender": + return poll_windows_defender(config) + body = config.get("body") + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = _headers(config) + if data is not None: + headers.setdefault("Content-Type", "application/json") + request = urllib.request.Request( + str(config["url"]), + data=data, + headers=headers, + method=str(config.get("method", "POST" if data is not None else "GET")), + ) + with urllib.request.urlopen(request, timeout=float(config.get("timeout_seconds", 5))) as response: + raw = response.read(4_194_304).decode("utf-8") + try: + document = json.loads(raw) + except json.JSONDecodeError: + rows = [json.loads(line) for line in raw.splitlines() if line.strip()] + document = [row.get("result", row) if isinstance(row, dict) else row for row in rows] + records = _path(document, config.get("records_path"), document) + if isinstance(records, dict): + records = [records] + if not isinstance(records, list): + raise ValueError(f"{config.get('name', 'integration')}: records_path did not resolve to a list") + + mapping = {**DEFAULT_SEVERITY, **{str(k).lower(): int(v) for k, v in config.get("severity_map", {}).items()}} + payloads: list[dict[str, Any]] = [] + for record in records[: int(config.get("max_records", 200))]: + if not isinstance(record, dict): + continue + excluded = False + for field, values in config.get("exclude_values", {}).items(): + actual = str(_path(record, str(field), "")).lower() + if actual in {str(value).lower() for value in values}: + excluded = True + break + if excluded: + continue + raw_score = _path(record, config.get("score_path")) + severity = str(_path(record, config.get("severity_path"), "")).lower() + score = int(float(raw_score)) if raw_score is not None else mapping.get(severity, 0) + scope = {str(k): str(_path(record, str(v), "")) for k, v in config.get("scope_paths", {}).items()} + scope = {key: value for key, value in scope.items() if value} + payloads.append({ + "source": config.get("name", "unknown"), + "event_id": _path(record, config.get("id_path")), + "score": score, + "confidence": float(config.get("confidence", 0.9)), + "detail": _path(record, config.get("detail_path"), severity or "external security event"), + "ttl_seconds": int(config.get("ttl_seconds", 300)), + "scope": scope, + }) + return ingest(payloads) if payloads else 0 + + +def poll_windows_defender(config: dict[str, Any]) -> int: + script = ( + "$status=Get-MpComputerStatus;" + "$threats=@(Get-MpThreatDetection -ErrorAction SilentlyContinue | Select-Object -First 100);" + "[pscustomobject]@{healthy=($status.AntivirusEnabled -and $status.RealTimeProtectionEnabled);" + "age=[math]::Round(((Get-Date)-$status.AntivirusSignatureLastUpdated).TotalHours,1);threats=$threats}" + "| ConvertTo-Json -Depth 5 -Compress" + ) + result = subprocess.run( + ["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", script], + capture_output=True, + text=True, + timeout=float(config.get("timeout_seconds", 8)), + check=True, + ) + document = json.loads(result.stdout) + payloads: list[dict[str, Any]] = [] + if not document.get("healthy", False): + payloads.append({ + "source": config.get("name", "windows-defender"), + "event_id": "protection-disabled", + "score": 85, + "confidence": 1.0, + "detail": "Microsoft Defender antivirus or real-time protection is disabled.", + "ttl_seconds": int(config.get("ttl_seconds", 120)), + }) + if float(document.get("age", 0)) > 48: + payloads.append({ + "source": config.get("name", "windows-defender"), + "event_id": "stale-signatures", + "score": 55, + "confidence": 0.95, + "detail": f"Microsoft Defender signatures are {document['age']} hours old.", + "ttl_seconds": int(config.get("ttl_seconds", 120)), + }) + threats = document.get("threats") or [] + if isinstance(threats, dict): + threats = [threats] + for index, threat in enumerate(threats): + payloads.append({ + "source": config.get("name", "windows-defender"), + "event_id": threat.get("ThreatID") or f"threat-{index}", + "score": 95, + "confidence": 1.0, + "detail": str(threat.get("Resources") or "Microsoft Defender reported an active/recent threat."), + "ttl_seconds": int(config.get("threat_ttl_seconds", 3600)), + }) + return ingest(payloads) if payloads else 0 + + +def load_config(path: Path) -> list[dict[str, Any]]: + document = json.loads(path.read_text(encoding="utf-8")) + integrations = document.get("integrations", []) + if not isinstance(integrations, list): + raise ValueError("integrations must be a list") + return [item for item in integrations if isinstance(item, dict) and item.get("enabled", True)] + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Poll AV/EDR/SIEM JSON APIs into Intent Gate") + parser.add_argument("config", type=Path) + parser.add_argument("--once", action="store_true") + parser.add_argument("--interval", type=float, default=30.0) + args = parser.parse_args(argv) + while True: + for integration in load_config(args.config): + try: + accepted = poll_integration(integration) + print(f"{integration.get('name', 'integration')}: accepted {accepted} signal(s)") + except Exception as exc: # collector failures must not stop command execution + print(f"{integration.get('name', 'integration')}: {type(exc).__name__}: {exc}") + if args.once: + return 0 + time.sleep(max(1.0, args.interval)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/intentgate/context.py b/src/intentgate/context.py new file mode 100644 index 0000000..8871789 --- /dev/null +++ b/src/intentgate/context.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import json +import os +import getpass +import socket +import subprocess +from pathlib import Path + +from .models import CommandContext +from .provenance import read_cached_scan +from .integrations import read_posture +from .behavior import assess_anomaly +from .privilege import detect_privilege + + +STATE_DIR = Path(os.environ.get("UIG_STATE_DIR", Path.home() / ".intentgate")) +HISTORY_FILE = STATE_DIR / "history.jsonl" + + +def _git(cwd: Path, *args: str) -> str | None: + try: + result = subprocess.run( + ["git", *args], cwd=cwd, capture_output=True, text=True, timeout=0.08 + ) + except (OSError, subprocess.TimeoutExpired): + return None + value = result.stdout.strip() + return value if result.returncode == 0 and value else None + + +def _find_git_root(cwd: Path) -> Path | None: + for candidate in (cwd, *cwd.parents): + if (candidate / ".git").exists(): + return candidate + return None + + +def recent_commands(limit: int = 8) -> tuple[str, ...]: + try: + lines = HISTORY_FILE.read_text(encoding="utf-8").splitlines()[-limit:] + except OSError: + return () + items: list[str] = [] + for line in lines: + try: + items.append(str(json.loads(line)["command"])) + except (json.JSONDecodeError, KeyError, TypeError): + continue + return tuple(items) + + +def project_provenance_signals(cwd: Path) -> tuple[str, ...]: + """Cheap signals only. A future background analyzer can populate richer facts.""" + markers = { + "package.json": "node-project", + "pyproject.toml": "python-project", + "requirements.txt": "python-project", + "Cargo.toml": "rust-project", + "go.mod": "go-project", + ".openai": "generated-app-metadata", + ".cursor": "ai-editor-metadata", + ".windsurf": "ai-editor-metadata", + } + signals = {label for name, label in markers.items() if (cwd / name).exists()} + has_tests = any((cwd / name).exists() for name in ("tests", "test", "__tests__")) + if not has_tests and signals & {"node-project", "python-project", "rust-project", "go-project"}: + signals.add("no-obvious-tests") + return tuple(sorted(signals)) + + +def collect_context(argv: list[str], purpose: str | None = None, cwd: str | None = None) -> CommandContext: + path = Path(cwd or os.getcwd()).resolve() + root_path = _find_git_root(path) + git_root = str(root_path) if root_path else None + branch = _git(path, "branch", "--show-current") if root_path else None + dirty = bool(_git(path, "status", "--porcelain")) if root_path else False + project_root = root_path or path + provenance = read_cached_scan(project_root) + user_name = getpass.getuser() + posture = read_posture( + device=socket.gethostname(), + user=user_name, + project=str(project_root), + ) + privilege = detect_privilege() + anomaly = assess_anomaly(" ".join(argv), user_name=user_name) + return CommandContext( + cwd=str(path), + command=" ".join(argv), + argv=tuple(argv), + purpose=purpose, + recent_commands=recent_commands(), + git_root=git_root, + git_branch=branch, + git_dirty=dirty, + project_signals=project_provenance_signals(project_root), + provenance_risk=int(provenance.get("risk_score", 0)), + provenance_details=tuple(str(item) for item in provenance.get("details", ())), + external_risk=int(posture.get("risk_score", 0)), + external_sources=tuple(str(item) for item in posture.get("sources", ())), + external_details=tuple(str(item) for item in posture.get("details", ())), + user_name=user_name, + is_root=privilege.is_root, + is_admin=privilege.is_admin, + privilege_level=privilege.level, + anomaly_score=int(anomaly.get("score", 0)), + anomaly_details=tuple(str(item) for item in anomaly.get("details", ())), + ) diff --git a/src/intentgate/engine.py b/src/intentgate/engine.py new file mode 100644 index 0000000..8b10071 --- /dev/null +++ b/src/intentgate/engine.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +import hashlib +import time + +from .models import Assessment, CommandContext, Decision +from .rules import evaluate_rules + + +def assess(ctx: CommandContext) -> Assessment: + started = time.perf_counter_ns() + signals = evaluate_rules(ctx) + score = max(0, min(100, sum(signal.score for signal in signals))) + if score >= 85: + decision = Decision.BLOCK + elif score >= 40: + decision = Decision.REVIEW + else: + decision = Decision.ALLOW + fingerprint = hashlib.blake2s(ctx.command.encode("utf-8"), digest_size=8).hexdigest() + elapsed = (time.perf_counter_ns() - started) / 1_000_000 + return Assessment(decision, score, signals, elapsed, fingerprint) + diff --git a/src/intentgate/integrations.py b/src/intentgate/integrations.py new file mode 100644 index 0000000..0f60f66 --- /dev/null +++ b/src/intentgate/integrations.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import hashlib +import json +import os +import time +from pathlib import Path +from typing import Any + + +MAX_SIGNALS = 512 + + +def state_dir() -> Path: + return Path(os.environ.get("UIG_STATE_DIR", Path.home() / ".intentgate")) + + +def signal_store_path() -> Path: + return state_dir() / "external-signals.json" + + +def _load() -> dict[str, Any]: + try: + data = json.loads(signal_store_path().read_text(encoding="utf-8")) + return data if isinstance(data, dict) else {"signals": []} + except (OSError, json.JSONDecodeError, TypeError): + return {"schema_version": 1, "updated_at": 0, "signals": []} + + +def _write(data: dict[str, Any]) -> None: + path = signal_store_path() + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(".tmp") + temporary.write_text(json.dumps(data, separators=(",", ":")), encoding="utf-8") + os.replace(temporary, path) + + +def normalize_signal(payload: dict[str, Any]) -> dict[str, Any]: + now = time.time() + source = str(payload.get("source", "unknown"))[:80] + event_id = str(payload.get("event_id") or payload.get("id") or "")[:160] + if not event_id: + digest = hashlib.blake2s(json.dumps(payload, sort_keys=True).encode("utf-8"), digest_size=10).hexdigest() + event_id = digest + score = max(0, min(100, int(float(payload.get("score", 0))))) + confidence = max(0.0, min(1.0, float(payload.get("confidence", 1.0)))) + ttl = max(1, min(604_800, int(payload.get("ttl_seconds", 300)))) + scope = payload.get("scope") if isinstance(payload.get("scope"), dict) else {} + return { + "source": source, + "event_id": event_id, + "score": score, + "confidence": confidence, + "detail": str(payload.get("detail", ""))[:500], + "observed_at": float(payload.get("observed_at", now)), + "expires_at": now + ttl, + "scope": {str(k): str(v) for k, v in scope.items() if k in {"device", "user", "project"}}, + } + + +def ingest(payloads: list[dict[str, Any]]) -> int: + now = time.time() + data = _load() + current = { + (str(item.get("source")), str(item.get("event_id"))): item + for item in data.get("signals", []) + if float(item.get("expires_at", 0)) > now + } + for payload in payloads: + item = normalize_signal(payload) + current[(item["source"], item["event_id"])] = item + signals = sorted(current.values(), key=lambda item: float(item["observed_at"]), reverse=True)[:MAX_SIGNALS] + _write({"schema_version": 1, "updated_at": now, "signals": signals}) + return len(payloads) + + +def _matches_scope(item_scope: dict[str, Any], requested: dict[str, str | None]) -> bool: + for key, expected in item_scope.items(): + actual = requested.get(key) + if expected and actual and str(expected).lower() != str(actual).lower(): + return False + return True + + +def read_posture(*, device: str | None = None, user: str | None = None, project: str | None = None) -> dict[str, Any]: + now = time.time() + requested = {"device": device, "user": user, "project": project} + active = [ + item for item in _load().get("signals", []) + if float(item.get("expires_at", 0)) > now and _matches_scope(item.get("scope", {}), requested) + ] + weighted = sorted( + [(float(item.get("score", 0)) * float(item.get("confidence", 1)), item) for item in active], + reverse=True, + key=lambda pair: pair[0], + ) + if not weighted: + return {"risk_score": 0, "sources": [], "details": [], "active_signals": 0} + highest = weighted[0][0] + corroboration = min(20.0, sum(value * 0.15 for value, _ in weighted[1:])) + risk = min(100, round(highest + corroboration)) + sources = sorted({str(item.get("source", "unknown")) for _, item in weighted}) + details = [str(item.get("detail", "")) for _, item in weighted[:5] if item.get("detail")] + return {"risk_score": risk, "sources": sources, "details": details, "active_signals": len(weighted)} + + +def source_postures() -> dict[str, int]: + now = time.time() + result: dict[str, int] = {} + for item in _load().get("signals", []): + if float(item.get("expires_at", 0)) <= now: + continue + source = str(item.get("source", "unknown")) + effective = round(float(item.get("score", 0)) * float(item.get("confidence", 1))) + result[source] = max(result.get(source, 0), effective) + return result diff --git a/src/intentgate/models.py b/src/intentgate/models.py new file mode 100644 index 0000000..14f4585 --- /dev/null +++ b/src/intentgate/models.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +from enum import Enum +from typing import Any + + +class Decision(str, Enum): + ALLOW = "allow" + REVIEW = "review" + BLOCK = "block" + + +@dataclass(frozen=True) +class Signal: + name: str + score: int + detail: str + + +@dataclass(frozen=True) +class CommandContext: + cwd: str + command: str + argv: tuple[str, ...] + purpose: str | None = None + recent_commands: tuple[str, ...] = () + git_root: str | None = None + git_branch: str | None = None + git_dirty: bool = False + project_signals: tuple[str, ...] = () + provenance_risk: int = 0 + provenance_details: tuple[str, ...] = () + external_risk: int = 0 + external_sources: tuple[str, ...] = () + external_details: tuple[str, ...] = () + user_name: str = "unknown" + is_root: bool = False + is_admin: bool = False + privilege_level: str = "standard" + anomaly_score: int = 0 + anomaly_details: tuple[str, ...] = () + + +@dataclass +class Assessment: + decision: Decision + risk_score: int + signals: list[Signal] = field(default_factory=list) + latency_ms: float = 0.0 + command_fingerprint: str = "" + + def to_dict(self) -> dict[str, Any]: + result = asdict(self) + result["decision"] = self.decision.value + return result diff --git a/src/intentgate/notifier.py b/src/intentgate/notifier.py new file mode 100644 index 0000000..80753be --- /dev/null +++ b/src/intentgate/notifier.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import argparse +import json +import os +import time +import urllib.request +from pathlib import Path +from typing import Any + +from .reporting import reports_dir + + +def _message(report: dict[str, Any]) -> str: + subject = report.get("subject", {}) + signals = ", ".join(item.get("name", "unknown") for item in report.get("signals", [])[:8]) + return ( + f"Intent Gate risk report: {report.get('decision', 'unknown').upper()} " + f"risk={report.get('risk_score', 0)}/100; user={subject.get('user', 'unknown')} " + f"privilege={subject.get('privilege_level', 'unknown')}; command={report.get('command', '')}; " + f"signals={signals}; report={report.get('report_id', '')}" + ) + + +def _payload(report: dict[str, Any], style: str) -> dict[str, Any]: + message = _message(report) + if style == "slack": + return {"text": message} + if style == "teams": + return {"type": "message", "attachments": [{"contentType": "application/vnd.microsoft.card.adaptive", "content": {"type": "AdaptiveCard", "version": "1.4", "body": [{"type": "TextBlock", "wrap": True, "text": message}]}}]} + return {"event": "intentgate.manager_risk_report", "summary": message, "report": report} + + +def deliver(path: Path, url: str, token: str | None, style: str) -> None: + report = json.loads(path.read_text(encoding="utf-8")) + body = json.dumps(_payload(report, style)).encode("utf-8") + headers = {"Content-Type": "application/json", "User-Agent": "IntentGate/0.3"} + if token: + headers["Authorization"] = f"Bearer {token}" + request = urllib.request.Request(url, data=body, headers=headers, method="POST") + with urllib.request.urlopen(request, timeout=5) as response: + if not 200 <= response.status < 300: + raise RuntimeError(f"manager webhook returned HTTP {response.status}") + path.unlink() + + +def run_once(url: str, token: str | None, style: str) -> tuple[int, int]: + sent = failed = 0 + for path in sorted(reports_dir().glob("*.json"))[:100]: + try: + deliver(path, url, token, style) + sent += 1 + except Exception as exc: + failed += 1 + print(f"{path.name}: {type(exc).__name__}: {exc}") + return sent, failed + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Deliver queued Intent Gate risk reports to a manager webhook") + parser.add_argument("--once", action="store_true") + parser.add_argument("--interval", type=float, default=10.0) + parser.add_argument("--style", choices=("generic", "slack", "teams"), default=os.environ.get("UIG_MANAGER_WEBHOOK_STYLE", "generic")) + args = parser.parse_args(argv) + url = os.environ.get("UIG_MANAGER_WEBHOOK_URL", "") + if not url: + print("UIG_MANAGER_WEBHOOK_URL is required; queued reports were not sent.") + return 2 + token = os.environ.get("UIG_MANAGER_WEBHOOK_TOKEN") + while True: + sent, failed = run_once(url, token, args.style) + print(f"manager notifier: sent={sent} failed={failed}") + if args.once: + return 0 if failed == 0 else 1 + time.sleep(max(1.0, args.interval)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/intentgate/privilege.py b/src/intentgate/privilege.py new file mode 100644 index 0000000..586fcee --- /dev/null +++ b/src/intentgate/privilege.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +import ctypes +import os +from dataclasses import dataclass + + +@dataclass(frozen=True) +class PrivilegeContext: + is_root: bool + is_admin: bool + level: str + + +def detect_privilege() -> PrivilegeContext: + if os.name == "nt": + try: + elevated = bool(ctypes.windll.shell32.IsUserAnAdmin()) + except (AttributeError, OSError): + elevated = False + return PrivilegeContext(False, elevated, "administrator" if elevated else "standard") + try: + root = os.geteuid() == 0 + except AttributeError: + root = False + if root: + return PrivilegeContext(True, True, "root") + try: + import grp + admin_groups = {"sudo", "wheel", "admin"} + group_names = {grp.getgrgid(group_id).gr_name for group_id in os.getgroups()} + admin = bool(group_names & admin_groups) + except (KeyError, OSError): + admin = False + return PrivilegeContext(False, admin, "administrator-capable" if admin else "standard") diff --git a/src/intentgate/provenance.py b/src/intentgate/provenance.py new file mode 100644 index 0000000..2329648 --- /dev/null +++ b/src/intentgate/provenance.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import time +from pathlib import Path +from typing import Any + + +SOURCE_EXTENSIONS = {".py", ".js", ".jsx", ".ts", ".tsx", ".go", ".rs", ".java", ".cs", ".ps1", ".sh"} +IGNORED_DIRS = {".git", ".venv", "venv", "node_modules", "dist", "build", "coverage", "vendor"} +AI_MARKERS = re.compile(r"(?:generated\s+by|chatgpt|claude|copilot|cursor\s+ai|as\s+an\s+ai)", re.IGNORECASE) +PLACEHOLDERS = re.compile(r"\b(?:TODO|FIXME|HACK|placeholder|not\s+implemented)\b", re.IGNORECASE) +SWALLOWED_ERRORS = re.compile(r"except\s+(?:Exception|BaseException)?\s*:\s*(?:#.*\n\s*)?pass\b", re.IGNORECASE) + + +def _state_dir() -> Path: + return Path(os.environ.get("UIG_STATE_DIR", Path.home() / ".intentgate")) + + +def _cache_path(root: Path) -> Path: + key = hashlib.blake2s(str(root.resolve()).lower().encode("utf-8"), digest_size=10).hexdigest() + return _state_dir() / "provenance" / f"{key}.json" + + +def _source_files(root: Path, limit: int = 250): + count = 0 + for current, dirs, files in os.walk(root): + dirs[:] = [name for name in dirs if name not in IGNORED_DIRS and not name.startswith(".")] + for name in files: + path = Path(current) / name + if path.suffix.lower() not in SOURCE_EXTENSIONS: + continue + yield path + count += 1 + if count >= limit: + return + + +def scan_project(root: Path, limit: int = 250) -> dict[str, Any]: + root = root.resolve() + files = list(_source_files(root, limit)) + test_files = [path for path in files if "test" in path.name.lower() or "tests" in path.parts] + ai_markers = placeholders = swallowed = large_files = unreadable = 0 + lines = 0 + for path in files: + try: + text = path.read_text(encoding="utf-8", errors="ignore")[:300_000] + except OSError: + unreadable += 1 + continue + line_count = text.count("\n") + 1 + lines += line_count + large_files += int(line_count > 1200) + ai_markers += len(AI_MARKERS.findall(text)) + placeholders += len(PLACEHOLDERS.findall(text)) + swallowed += len(SWALLOWED_ERRORS.findall(text)) + + score = 0 + details: list[str] = [] + if files and not test_files: + score += 25 + details.append("No test source files were found in the bounded scan.") + if ai_markers: + score += min(15, ai_markers * 3) + details.append(f"Found {ai_markers} explicit AI-generation/provenance marker(s).") + if placeholders >= 8: + score += min(20, placeholders) + details.append(f"Found {placeholders} placeholder or unfinished-code marker(s).") + if swallowed: + score += min(20, swallowed * 5) + details.append(f"Found {swallowed} broad exception handler(s) that silently discard errors.") + if large_files: + score += min(15, large_files * 5) + details.append(f"Found {large_files} source file(s) over 1,200 lines.") + if not files: + details.append("No supported source files were found; provenance confidence is unknown.") + + result = { + "schema_version": 1, + "root": str(root), + "scanned_at": time.time(), + "risk_score": min(100, score), + "details": details, + "metrics": { + "source_files": len(files), + "test_files": len(test_files), + "lines_sampled": lines, + "ai_markers": ai_markers, + "placeholders": placeholders, + "swallowed_errors": swallowed, + "large_files": large_files, + "unreadable_files": unreadable, + "file_limit": limit, + }, + } + cache = _cache_path(root) + cache.parent.mkdir(parents=True, exist_ok=True) + cache.write_text(json.dumps(result, indent=2), encoding="utf-8") + return result + + +def read_cached_scan(root: Path, max_age_seconds: int = 86_400) -> dict[str, Any]: + try: + result = json.loads(_cache_path(root).read_text(encoding="utf-8")) + if time.time() - float(result["scanned_at"]) > max_age_seconds: + return {} + return result + except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError): + return {} + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Cache bounded project provenance signals for Intent Gate") + parser.add_argument("path", nargs="?", default=".") + parser.add_argument("--limit", type=int, default=250) + args = parser.parse_args(argv) + result = scan_project(Path(args.path), max(1, min(args.limit, 2_000))) + print(json.dumps(result, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/intentgate/reporting.py b/src/intentgate/reporting.py new file mode 100644 index 0000000..2ce6171 --- /dev/null +++ b/src/intentgate/reporting.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import hashlib +import json +import os +import re +import time +import uuid +from pathlib import Path + +from .integrations import state_dir +from .models import Assessment, CommandContext, Decision + + +SECRET_PATTERNS = ( + re.compile(r"(?i)(--?(?:password|passwd|token|secret|api[-_]?key|authorization)\s*[= ]\s*)(\S+)"), + re.compile(r"(?i)(https?://[^\s:/]+:)([^@\s]+)(@)"), + re.compile(r"(?i)(bearer\s+)([a-z0-9._~+/-]+=*)"), + re.compile(r"(?i)((?:aws_secret_access_key|client_secret|private_key)\s*[=:]\s*)(\S+)"), +) + + +def reports_dir() -> Path: + return state_dir() / "manager-reports" + + +def redact_command(command: str) -> str: + redacted = command + for pattern in SECRET_PATTERNS: + if pattern.groups == 3: + redacted = pattern.sub(r"\1[REDACTED]\3", redacted) + else: + redacted = pattern.sub(r"\1[REDACTED]", redacted) + return redacted[:2000] + + +def should_escalate(ctx: CommandContext, result: Assessment) -> bool: + threshold = max(1, min(100, int(os.environ.get("UIG_MANAGER_REPORT_THRESHOLD", "70")))) + privileged_anomaly = (ctx.is_root or ctx.is_admin) and ctx.anomaly_score >= 20 and result.risk_score >= 40 + unusual = ctx.anomaly_score >= 25 + return result.risk_score >= threshold or result.decision is Decision.BLOCK or privileged_anomaly or unusual + + +def queue_manager_report(ctx: CommandContext, result: Assessment, executed: bool, exit_code: int | None) -> Path | None: + if not should_escalate(ctx, result): + return None + directory = reports_dir() + directory.mkdir(parents=True, exist_ok=True) + report_id = f"{int(time.time() * 1000)}-{uuid.uuid4().hex[:10]}" + command = redact_command(ctx.command) + report = { + "schema_version": 1, + "report_id": report_id, + "created_at": time.time(), + "manager": os.environ.get("UIG_MANAGER_ID", "unconfigured"), + "subject": { + "user": ctx.user_name, + "privilege_level": ctx.privilege_level, + "is_root": ctx.is_root, + "is_admin": ctx.is_admin, + "cwd": ctx.cwd, + }, + "command": command, + "command_hash": hashlib.sha256(ctx.command.encode("utf-8")).hexdigest(), + "purpose": ctx.purpose, + "decision": result.decision.value, + "risk_score": result.risk_score, + "anomaly_score": ctx.anomaly_score, + "signals": [ + {"name": signal.name, "score": signal.score, "detail": signal.detail} + for signal in result.signals if signal.score > 0 + ], + "external_sources": list(ctx.external_sources), + "executed": executed, + "exit_code": exit_code, + } + target = directory / f"{report_id}.json" + temporary = directory / f".{report_id}.tmp" + temporary.write_text(json.dumps(report, indent=2), encoding="utf-8") + os.replace(temporary, target) + return target + + +def pending_report_count() -> int: + try: + return sum(1 for _ in reports_dir().glob("*.json")) + except OSError: + return 0 diff --git a/src/intentgate/rules.py b/src/intentgate/rules.py new file mode 100644 index 0000000..c2d7a72 --- /dev/null +++ b/src/intentgate/rules.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import re +from pathlib import Path + +from .catalog import match_destructive_actions +from .models import CommandContext, Signal + + +DESTRUCTIVE = re.compile( + r"(?:^|\s)(?:rm\s+-[^\n]*r|rmdir\s+/s|del\s+/[sq]|remove-item\b[^\n]*(?:-recurse|-force)|" + r"format(?:\.com)?\b|mkfs\b|diskpart\b|drop\s+(?:database|table)\b|truncate\s+table\b)", + re.IGNORECASE, +) +PRIVILEGE = re.compile(r"(?:^|\s)(?:sudo|runas|doas|set-executionpolicy|takeown|icacls)\b", re.IGNORECASE) +NETWORK_EXEC = re.compile( + r"(?:curl|wget|irm|invoke-restmethod|iwr|invoke-webrequest)[^\n|;]*(?:\||;|&&)\s*(?:sh|bash|pwsh|powershell|python|node)", + re.IGNORECASE, +) +SECRET_ACCESS = re.compile(r"(?:\.env\b|id_rsa|credentials|secrets?\.|token\b|keychain)", re.IGNORECASE) +EXFIL = re.compile(r"(?:curl|wget|scp|rsync|invoke-restmethod|invoke-webrequest)\b", re.IGNORECASE) +OBFUSCATION = re.compile(r"(?:frombase64string|-enc(?:odedcommand)?\b|base64\s+-d|eval\s*\(|iex\b)", re.IGNORECASE) +PUBLISH = re.compile(r"(?:git\s+push|npm\s+publish|twine\s+upload|docker\s+push|kubectl\s+apply|terraform\s+apply)", re.IGNORECASE) +READ_ONLY = re.compile( + r"^(?:git\s+(?:status|diff|log|show|branch)|(?:ls|dir|pwd|whoami|where|which|type|get-childitem|get-content|rg)\b|" + r"(?:python|node|git|docker|npm|pip)\s+--version)", + re.IGNORECASE, +) + + +def _targets_broad_scope(command: str, cwd: str) -> bool: + normalized = command.replace("\\", "/").lower() + broad = (" / " in f" {normalized} ", " c:/" in normalized, " $home" in normalized, " ~" in normalized) + if any(broad): + return True + # Parent traversal plus destructive operation is a blast-radius hint. + return "../" in normalized or "..\\" in command.lower() + + +def evaluate_rules(ctx: CommandContext) -> list[Signal]: + command = ctx.command.strip() + signals: list[Signal] = [] + if not command: + return [Signal("empty-command", 100, "No executable command was supplied.")] + if ctx.is_root: + signals.append(Signal("privilege-context", 0, "Current process is running as Linux root.")) + elif ctx.is_admin: + signals.append(Signal("privilege-context", 0, f"Current user/process privilege is {ctx.privilege_level}.")) + if READ_ONLY.search(command): + signals.append(Signal("read-only", -25, "Command matches a common read-only operation.")) + destructive_actions = match_destructive_actions(command) + if destructive_actions: + highest = max(item.score for item in destructive_actions) + names = ", ".join(item.identifier for item in destructive_actions[:3]) + signals.append(Signal("destructive-action", highest, f"Matched destructive catalog action(s): {names}.")) + elif DESTRUCTIVE.search(command): + signals.append(Signal("destructive", 55, "Command can delete or irreversibly alter data.")) + if PRIVILEGE.search(command): + signals.append(Signal("privilege", 25, "Command requests or changes elevated privileges.")) + if NETWORK_EXEC.search(command): + signals.append(Signal("network-to-execution", 80, "Downloaded content appears to flow directly into an interpreter.")) + if OBFUSCATION.search(command): + signals.append(Signal("obfuscation", 45, "Command contains an encoding or dynamic-execution pattern.")) + if SECRET_ACCESS.search(command): + signals.append(Signal("secret-access", 25, "Command references likely credentials or secrets.")) + if SECRET_ACCESS.search(command) and EXFIL.search(command): + signals.append(Signal("possible-exfiltration", 55, "Network transfer and secret access occur together.")) + if PUBLISH.search(command): + signals.append(Signal("external-side-effect", 30, "Command can publish or modify external infrastructure.")) + if DESTRUCTIVE.search(command) and _targets_broad_scope(command, ctx.cwd): + signals.append(Signal("large-blast-radius", 45, "Destructive command appears to target a broad or parent scope.")) + if ctx.git_dirty and (DESTRUCTIVE.search(command) or PUBLISH.search(command)): + signals.append(Signal("dirty-worktree", 15, "Risky operation is being run with uncommitted changes.")) + if "ai-editor-metadata" in ctx.project_signals or "generated-app-metadata" in ctx.project_signals: + signals.append(Signal("ai-assisted-project", 5, "Project contains AI-tool metadata; provenance confidence is reduced slightly.")) + if "no-obvious-tests" in ctx.project_signals and PUBLISH.search(command): + signals.append(Signal("untested-publish", 20, "No obvious test directory was found before a publish/deploy operation.")) + if ctx.provenance_risk >= 60 and (PUBLISH.search(command) or PRIVILEGE.search(command)): + detail = ctx.provenance_details[0] if ctx.provenance_details else "Cached project provenance risk is high." + signals.append(Signal("provenance-risk", 25, detail)) + if ctx.external_risk >= 90: + sources = ", ".join(ctx.external_sources) or "external security tools" + signals.append(Signal("critical-security-posture", 55, f"{sources} report critical correlated risk ({ctx.external_risk}/100).")) + elif ctx.external_risk >= 70: + sources = ", ".join(ctx.external_sources) or "external security tools" + signals.append(Signal("high-security-posture", 30, f"{sources} report high correlated risk ({ctx.external_risk}/100).")) + elif ctx.external_risk >= 40: + sources = ", ".join(ctx.external_sources) or "external security tools" + signals.append(Signal("elevated-security-posture", 15, f"{sources} report elevated correlated risk ({ctx.external_risk}/100).")) + risky_base = sum(signal.score for signal in signals if signal.score > 0) + if ctx.is_root and risky_base >= 30: + signals.append(Signal("root-risk-amplifier", 30, "A risky command is being executed as Linux root.")) + elif ctx.is_admin and risky_base >= 30: + signals.append(Signal("admin-risk-amplifier", 25, "A risky command is being executed from an elevated Windows administrator session.")) + if ctx.anomaly_score >= 20 and risky_base >= 25: + detail = ctx.anomaly_details[0] if ctx.anomaly_details else "Command is unusual for this user's recent baseline." + signals.append(Signal("behavioral-anomaly", min(30, ctx.anomaly_score), detail)) + if ctx.purpose: + purpose_terms = {w for w in re.findall(r"[a-z0-9]+", ctx.purpose.lower()) if len(w) > 3} + command_terms = set(re.findall(r"[a-z0-9]+", command.lower())) + if purpose_terms and not (purpose_terms & command_terms) and sum(s.score for s in signals) >= 30: + signals.append(Signal("purpose-mismatch", 20, "Declared purpose has little lexical overlap with a risky command.")) + elif sum(s.score for s in signals) >= 30: + signals.append(Signal("missing-purpose", 10, "No purpose was supplied for a risky operation.")) + if ctx.recent_commands: + recent = "\n".join(ctx.recent_commands[-4:]) + if SECRET_ACCESS.search(recent) and EXFIL.search(command): + signals.append(Signal("sequence-risk", 35, "Recent secret access followed by network transfer increases exfiltration risk.")) + return signals diff --git a/src/intentgate/service.py b/src/intentgate/service.py new file mode 100644 index 0000000..9d962dc --- /dev/null +++ b/src/intentgate/service.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import argparse +import hmac +import json +import os +import re +import time +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlparse + +from .context import HISTORY_FILE +from .integrations import ingest, read_posture, source_postures +from .reporting import pending_report_count + + +STARTED_AT = time.time() + + +def _audit_metrics() -> tuple[dict[str, int], float, int]: + counts = {"allow": 0, "review": 0, "block": 0} + latencies: list[float] = [] + try: + lines = HISTORY_FILE.read_text(encoding="utf-8").splitlines()[-5000:] + except OSError: + return counts, 0.0, 0 + for line in lines: + try: + item = json.loads(line) + decision = str(item.get("decision", "")) + if decision in counts: + counts[decision] += 1 + if "latency_ms" in item: + latencies.append(float(item["latency_ms"])) + except (json.JSONDecodeError, TypeError, ValueError): + continue + average = sum(latencies) / len(latencies) if latencies else 0.0 + return counts, average, len(lines) + + +def prometheus_metrics() -> str: + posture = read_posture() + counts, average_latency, events = _audit_metrics() + lines = [ + "# HELP intentgate_security_posture Current aggregate external security risk score.", + "# TYPE intentgate_security_posture gauge", + f"intentgate_security_posture {posture['risk_score']}", + "# HELP intentgate_active_external_signals Number of non-expired external security signals.", + "# TYPE intentgate_active_external_signals gauge", + f"intentgate_active_external_signals {posture['active_signals']}", + "# HELP intentgate_decisions_total Gate decisions observed in the local audit history.", + "# TYPE intentgate_decisions_total counter", + ] + for decision, count in counts.items(): + lines.append(f'intentgate_decisions_total{{decision="{decision}"}} {count}') + lines.extend([ + "# HELP intentgate_audit_events Number of audit events currently included.", + "# TYPE intentgate_audit_events gauge", + f"intentgate_audit_events {events}", + "# HELP intentgate_decision_latency_ms Average recorded policy-engine latency.", + "# TYPE intentgate_decision_latency_ms gauge", + f"intentgate_decision_latency_ms {average_latency:.6f}", + "# HELP intentgate_uptime_seconds Integration service uptime.", + "# TYPE intentgate_uptime_seconds gauge", + f"intentgate_uptime_seconds {time.time() - STARTED_AT:.3f}", + "# HELP intentgate_pending_manager_reports Queued risk reports awaiting delivery.", + "# TYPE intentgate_pending_manager_reports gauge", + f"intentgate_pending_manager_reports {pending_report_count()}", + "# HELP intentgate_source_risk External security risk by normalized source.", + "# TYPE intentgate_source_risk gauge", + ]) + for source, score in source_postures().items(): + safe = re.sub(r"[^a-zA-Z0-9_.:-]", "_", source) + lines.append(f'intentgate_source_risk{{source="{safe}"}} {score}') + return "\n".join(lines) + "\n" + + +class Handler(BaseHTTPRequestHandler): + server_version = "IntentGate/0.1" + + def _json(self, status: int, body: object) -> None: + payload = json.dumps(body).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def _authorized(self) -> bool: + expected = os.environ.get("UIG_INGEST_TOKEN", "") + if not expected: + return self.client_address[0] in {"127.0.0.1", "::1"} + supplied = self.headers.get("Authorization", "").removeprefix("Bearer ") + return hmac.compare_digest(expected, supplied) + + def do_GET(self) -> None: + parsed = urlparse(self.path) + if parsed.path == "/healthz": + self._json(HTTPStatus.OK, {"status": "ok"}) + return + if parsed.path == "/v1/posture": + query = parse_qs(parsed.query) + self._json(HTTPStatus.OK, read_posture( + device=query.get("device", [None])[0], + user=query.get("user", [None])[0], + project=query.get("project", [None])[0], + )) + return + if parsed.path == "/metrics": + payload = prometheus_metrics().encode("utf-8") + self.send_response(HTTPStatus.OK) + self.send_header("Content-Type", "text/plain; version=0.0.4") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + return + self._json(HTTPStatus.NOT_FOUND, {"error": "not found"}) + + def do_POST(self) -> None: + if self.path not in {"/v1/signals", "/v1/events"}: + self._json(HTTPStatus.NOT_FOUND, {"error": "not found"}) + return + if not self._authorized(): + self._json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"}) + return + try: + length = min(int(self.headers.get("Content-Length", "0")), 1_048_576) + body = json.loads(self.rfile.read(length)) + items = body if isinstance(body, list) else [body] + if not all(isinstance(item, dict) for item in items): + raise ValueError("payload must be an object or list of objects") + accepted = ingest(items) + except (json.JSONDecodeError, TypeError, ValueError) as exc: + self._json(HTTPStatus.BAD_REQUEST, {"error": str(exc)}) + return + self._json(HTTPStatus.ACCEPTED, {"accepted": accepted}) + + def log_message(self, format: str, *args) -> None: + if os.environ.get("UIG_HTTP_LOG", "") == "1": + super().log_message(format, *args) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Intent Gate integration and metrics service") + parser.add_argument("--host", default=os.environ.get("UIG_HOST", "127.0.0.1")) + parser.add_argument("--port", type=int, default=int(os.environ.get("UIG_PORT", "8787"))) + args = parser.parse_args(argv) + server = ThreadingHTTPServer((args.host, args.port), Handler) + print(f"Intent Gate service listening on http://{args.host}:{args.port}") + try: + server.serve_forever() + except KeyboardInterrupt: + pass + finally: + server.server_close() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_engine.py b/tests/test_engine.py new file mode 100644 index 0000000..c2d3686 --- /dev/null +++ b/tests/test_engine.py @@ -0,0 +1,70 @@ +import unittest +import tempfile +import os +from pathlib import Path +from unittest.mock import patch + +from intentgate.context import collect_context +from intentgate.engine import assess +from intentgate.models import Decision +from intentgate.integrations import ingest, read_posture +from intentgate.models import CommandContext +from intentgate.provenance import scan_project + + +def check(command: list[str], purpose: str | None = None): + return assess(collect_context(command, purpose=purpose, cwd=".")) + + +class EngineTests(unittest.TestCase): + def test_read_only_command_is_allowed(self): + result = check(["git", "status"]) + self.assertIs(result.decision, Decision.ALLOW) + + def test_network_to_shell_is_blocked(self): + result = check(["curl", "https://example.test/install.sh", "|", "bash"]) + self.assertIs(result.decision, Decision.BLOCK) + self.assertIn("network-to-execution", {signal.name for signal in result.signals}) + + def test_recursive_delete_requires_review(self): + result = check(["Remove-Item", "-Recurse", "./build"]) + self.assertIn(result.decision, {Decision.REVIEW, Decision.BLOCK}) + + def test_broad_recursive_delete_is_blocked(self): + result = check(["Remove-Item", "-Recurse", "-Force", "C:\\"]) + self.assertIs(result.decision, Decision.BLOCK) + self.assertIn("large-blast-radius", {signal.name for signal in result.signals}) + + def test_publish_is_reviewed_without_purpose(self): + result = check(["git", "push"]) + self.assertIs(result.decision, Decision.REVIEW) + + def test_provenance_scan_reports_missing_tests(self): + with tempfile.TemporaryDirectory() as directory: + Path(directory, "app.py").write_text("print('hello')\n", encoding="utf-8") + result = scan_project(Path(directory)) + self.assertGreaterEqual(result["risk_score"], 25) + self.assertEqual(result["metrics"]["source_files"], 1) + + def test_external_security_score_is_correlated(self): + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, {"UIG_STATE_DIR": directory}): + ingest([ + {"source": "defender", "event_id": "a", "score": 80, "confidence": 1, "ttl_seconds": 60}, + {"source": "splunk", "event_id": "b", "score": 60, "confidence": 0.5, "ttl_seconds": 60}, + ]) + result = read_posture() + self.assertEqual(result["risk_score"], 84) + self.assertEqual(result["active_signals"], 2) + + def test_critical_external_posture_blocks_publish(self): + context = CommandContext( + cwd=".", command="git push", argv=("git", "push"), + external_risk=95, external_sources=("defender",), + ) + result = assess(context) + self.assertIs(result.decision, Decision.BLOCK) + self.assertIn("critical-security-posture", {signal.name for signal in result.signals}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_security_context.py b/tests/test_security_context.py new file mode 100644 index 0000000..43ee385 --- /dev/null +++ b/tests/test_security_context.py @@ -0,0 +1,92 @@ +import json +import os +import tempfile +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from unittest.mock import patch + +from intentgate.behavior import assess_anomaly +from intentgate.catalog import match_destructive_actions +from intentgate.engine import assess +from intentgate.models import CommandContext, Decision +from intentgate.reporting import queue_manager_report, redact_command +from intentgate.notifier import run_once + + +class SecurityContextTests(unittest.TestCase): + def test_linux_and_windows_destructive_catalog(self): + linux = {item.identifier for item in match_destructive_actions("rm -rf /srv/app")} + windows = {item.identifier for item in match_destructive_actions("vssadmin delete shadows /all")} + self.assertIn("recursive-delete-unix", linux) + self.assertIn("shadow-copy-delete", windows) + + def test_root_amplifies_risky_command(self): + context = CommandContext( + cwd="/srv/app", command="systemctl stop auditd", argv=("systemctl", "stop", "auditd"), + user_name="root", is_root=True, is_admin=True, privilege_level="root", + ) + result = assess(context) + names = {signal.name for signal in result.signals} + self.assertIs(result.decision, Decision.BLOCK) + self.assertIn("root-risk-amplifier", names) + self.assertIn("destructive-action", names) + + def test_behavioral_baseline_finds_unseen_family(self): + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, {"UIG_STATE_DIR": directory}): + history = Path(directory, "history.jsonl") + rows = [{"command": "git status", "user_name": "example-user"} for _ in range(25)] + history.write_text("\n".join(json.dumps(row) for row in rows), encoding="utf-8") + result = assess_anomaly("terraform destroy", user_name="example-user") + self.assertGreaterEqual(result["score"], 20) + + def test_manager_report_is_queued_and_secrets_are_redacted(self): + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, {"UIG_STATE_DIR": directory}): + context = CommandContext( + cwd=".", command="deploy --token synthetic-secret", argv=("deploy",), user_name="example-user", + is_admin=True, privilege_level="administrator", + ) + result = assess(CommandContext(cwd=".", command="format C:", argv=("format", "C:"), user_name="example-user")) + path = queue_manager_report(context, result, executed=False, exit_code=None) + self.assertIsNotNone(path) + report = json.loads(path.read_text(encoding="utf-8")) + self.assertNotIn("synthetic-secret", report["command"]) + self.assertIn("[REDACTED]", report["command"]) + + def test_redacts_url_credentials(self): + value = redact_command("curl https://example-user:synthetic-password@example.invalid/api") + self.assertNotIn("synthetic-password", value) + + def test_notifier_delivers_and_removes_report(self): + received = [] + + class Receiver(BaseHTTPRequestHandler): + def do_POST(self): + received.append(json.loads(self.rfile.read(int(self.headers["Content-Length"])))) + self.send_response(204) + self.end_headers() + + def log_message(self, format, *args): + pass + + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, {"UIG_STATE_DIR": directory}): + context = CommandContext(cwd=".", command="format C:", argv=("format", "C:"), user_name="example-user") + result = assess(context) + queued = queue_manager_report(context, result, executed=False, exit_code=None) + server = ThreadingHTTPServer(("127.0.0.1", 0), Receiver) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + sent, failed = run_once(f"http://127.0.0.1:{server.server_port}", None, "generic") + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + self.assertEqual((sent, failed), (1, 0)) + self.assertFalse(queued.exists()) + self.assertEqual(received[0]["event"], "intentgate.manager_risk_report") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..41d054b --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,51 @@ +import json +import os +import tempfile +import threading +import unittest +import urllib.request +from http.server import ThreadingHTTPServer +from unittest.mock import patch + +from intentgate.service import Handler + + +class ServiceTests(unittest.TestCase): + def test_ingest_posture_and_metrics(self): + with tempfile.TemporaryDirectory() as directory, patch.dict( + os.environ, + {"UIG_STATE_DIR": directory, "UIG_INGEST_TOKEN": "synthetic-test-token"}, + ): + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + base = f"http://127.0.0.1:{server.server_port}" + try: + payload = json.dumps({ + "source": "smoke-siem", + "event_id": "event-1", + "score": 88, + "confidence": 1, + "ttl_seconds": 60, + "detail": "test alert", + }).encode("utf-8") + request = urllib.request.Request( + f"{base}/v1/signals", + data=payload, + headers={"Authorization": "Bearer synthetic-test-token", "Content-Type": "application/json"}, + method="POST", + ) + accepted = json.loads(urllib.request.urlopen(request, timeout=2).read()) + posture = json.loads(urllib.request.urlopen(f"{base}/v1/posture", timeout=2).read()) + metrics = urllib.request.urlopen(f"{base}/metrics", timeout=2).read().decode("utf-8") + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + self.assertEqual(accepted["accepted"], 1) + self.assertEqual(posture["risk_score"], 88) + self.assertIn("intentgate_security_posture 88", metrics) + + +if __name__ == "__main__": + unittest.main()