> ## Documentation Index
> Fetch the complete documentation index at: https://attesta.kyberon.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# DevOps Pipeline Example

> Complete example of an AI agent managing infrastructure deployments with SOC 2 compliance, deployment gating, rollback protection, and database migration controls

This example builds a DevOps AI agent that manages service deployments, database migrations, firewall rules, and cluster scaling. A custom infrastructure domain profile (created by you using the domain profile framework) can enforce SOC 2 and ISO 27001 compliance by detecting production environment operations, requiring multi-party approval for deployments, and protecting against unverified rollbacks.

## What You Will Build

* A deployment agent that gates production releases with multi-party approval
* Database migration management with DBA sign-off
* Firewall rule changes with teach-back verification
* Rollback protection that prevents panic-driven emergency changes
* Both Python and TypeScript implementations using LangGraph

***

## Configuration

Create an `attesta.yaml` in your project root:

```yaml attesta.yaml theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
# ─── Domain Profile (optional) ───────────────────────────────────
# domain: my-domain  # Activate a registered domain profile

# ─── Policy ───────────────────────────────────────────────────────
policy:
  minimum_review_seconds:
    low: 0
    medium: 3
    high: 15
    critical: 45

  require_multi_party:
    critical: 2

  fail_mode: deny
  timeout_seconds: 300

# ─── Risk Scoring ─────────────────────────────────────────────────
risk:
  overrides:
    drop_production_database: critical
    disable_firewall: critical
    delete_all_backups: critical
    terminate_production_cluster: critical
    disable_mfa: critical

  amplifiers:
    - pattern: ".*production.*"
      boost: 0.3
    - pattern: ".*deploy.*"
      boost: 0.15
    - pattern: ".*delete.*"
      boost: 0.2
    - pattern: ".*firewall.*"
      boost: 0.15
    - pattern: ".*migrate.*"
      boost: 0.1

# ─── Trust Engine ─────────────────────────────────────────────────
trust:
  influence: 0.25
  ceiling: 0.85
  initial_score: 0.2
  decay_rate: 0.01

# ─── Audit ────────────────────────────────────────────────────────
audit:
  path: ".attesta/infra-audit.jsonl"
```

<Note>
  A custom infrastructure domain profile might set a `production_multiplier` of 1.6. A deployment scored at 0.55 (MEDIUM) in staging would then become 0.88 (CRITICAL) in production, triggering multi-party approval. This is intentional -- the same operation carries fundamentally different risk in production. See [Custom Domains](/domains/custom-domains) for how to create your own domain profile.
</Note>

***

## Python Implementation

<Steps>
  <Step title="Install dependencies">
    ```bash theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
    pip install attesta[all] langchain-openai langgraph
    ```
  </Step>

  <Step title="Define the infrastructure tools">
    Each function is gated by Attesta. A custom domain profile can be configured to detect deployment operations, database commands, firewall changes, and privilege escalation.

    ```python infra_agent.py theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
    from attesta import Attesta
    from attesta import AttestaDenied

    # Load configuration (optionally with a custom domain profile)
    attesta = Attesta.from_config("attesta.yaml")


    # ─── Low risk: read-only status checks ────────────────────────────
    @attesta.gate()
    def get_service_status(service: str, environment: str = "production") -> dict:
        """Check the health and deployment status of a service."""
        return {
            "service": service,
            "environment": environment,
            "status": "healthy",
            "version": "2.0.3",
            "instances": 4,
            "uptime_hours": 720,
            "last_deploy": "2026-02-10T14:30:00Z",
        }


    @attesta.gate()
    def list_recent_deployments(environment: str, limit: int = 10) -> dict:
        """List recent deployments to an environment."""
        return {
            "environment": environment,
            "deployments": [
                {"service": "api-gateway", "version": "2.0.3",
                 "deployed_at": "2026-02-10T14:30:00Z", "status": "active"},
                {"service": "auth-service", "version": "1.8.1",
                 "deployed_at": "2026-02-09T09:15:00Z", "status": "active"},
                {"service": "payment-api", "version": "3.2.0",
                 "deployed_at": "2026-02-08T16:45:00Z", "status": "active"},
            ],
        }


    # ─── Medium risk: staging deployments ─────────────────────────────
    @attesta.gate(risk_hints={"environment": "staging"})
    def deploy_to_staging(
        service: str, version: str, strategy: str = "rolling"
    ) -> dict:
        """Deploy a service to the staging environment.

        Uses the specified deployment strategy (rolling, blue-green,
        or canary). Staging deployments require confirmation but not
        multi-party approval.
        """
        return {
            "deployment_id": "DEP-STG-2026-001",
            "service": service,
            "version": version,
            "environment": "staging",
            "strategy": strategy,
            "status": "deployed",
            "instances_updated": 2,
        }


    # ─── High/Critical risk: production deployments ───────────────────
    @attesta.gate(risk_hints={"production": True})
    def deploy_to_production(
        service: str,
        version: str,
        strategy: str = "canary",
        canary_percentage: int = 10,
        change_ticket: str = "",
    ) -> dict:
        """Deploy a service to the production environment.

        Production deployments require approval from the SRE lead and
        release manager per your domain's escalation rules.
        A valid change ticket from the CAB is expected.
        """
        return {
            "deployment_id": "DEP-PROD-2026-042",
            "service": service,
            "version": version,
            "environment": "production",
            "strategy": strategy,
            "canary_percentage": canary_percentage,
            "change_ticket": change_ticket,
            "status": "canary_deployed",
            "healthy_instances": 1,
            "total_instances": 10,
        }


    # ─── High risk: rollback operations ───────────────────────────────
    @attesta.gate(risk_hints={"production": True, "emergency": True})
    def rollback_deployment(
        service: str,
        target_version: str,
        reason: str,
    ) -> dict:
        """Rollback a production service to a previous version.

        Rollbacks are emergency operations that bypass the normal
        deployment pipeline. They still require approval to prevent
        panic-driven changes that make incidents worse.
        """
        return {
            "rollback_id": "RBK-2026-007",
            "service": service,
            "from_version": "2.1.0",
            "to_version": target_version,
            "environment": "production",
            "reason": reason,
            "status": "rolled_back",
            "instances_reverted": 10,
        }


    # ─── Critical risk: database migrations ───────────────────────────
    @attesta.gate(
        risk_hints={"production": True, "affects_database": True}
    )
    def run_migration(
        migration_file: str,
        database: str,
        environment: str = "production",
    ) -> dict:
        """Run a database schema migration.

        Schema migrations in production require approval from the DBA
        and backend engineering lead per your domain's escalation
        rules. Migrations are irreversible -- always verify the
        migration file contents before approving.
        """
        return {
            "migration_id": "MIG-2026-015",
            "migration_file": migration_file,
            "database": database,
            "environment": environment,
            "tables_affected": ["users", "orders"],
            "status": "migration_applied",
            "duration_seconds": 12.4,
        }


    # ─── High risk: firewall rule changes ─────────────────────────────
    @attesta.gate(risk_hints={"security_boundary": True})
    def modify_firewall_rule(
        rule_id: str,
        action: str,
        protocol: str,
        port: int,
        cidr: str,
        direction: str = "ingress",
    ) -> dict:
        """Modify a firewall rule in the production VPC.

        Firewall changes affect the network security boundary. Your
        domain profile can require teach-back verification and
        notification of the security engineer and network admin.
        """
        return {
            "rule_id": rule_id,
            "action": action,
            "protocol": protocol,
            "port": port,
            "cidr": cidr,
            "direction": direction,
            "status": "rule_updated",
        }


    # ─── High risk: credential rotation ───────────────────────────────
    @attesta.gate(risk_hints={"security_critical": True})
    def rotate_credentials(
        service: str,
        credential_type: str,
        environment: str = "production",
    ) -> dict:
        """Rotate credentials for a production service.

        Credential rotation temporarily creates a window where both
        old and new credentials are valid. Requires verification to
        ensure dependent services are prepared.
        """
        return {
            "service": service,
            "credential_type": credential_type,
            "environment": environment,
            "rotation_id": "ROT-2026-003",
            "status": "rotated",
            "old_credential_expires_at": "2026-02-14T09:00:00Z",
        }
    ```
  </Step>

  <Step title="Build the agent loop with LangGraph">
    This agent uses LangGraph with Attesta inserted as a gate node between the agent and tool execution.

    ```python infra_loop.py theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
    import asyncio
    from langchain_openai import ChatOpenAI
    from langchain_core.tools import StructuredTool
    from langgraph.graph import StateGraph, MessagesState
    from langgraph.prebuilt import ToolNode

    from attesta import Attesta
    from attesta.integrations.langchain import (
        AttestaToolWrapper,
        attesta_node,
    )

    from infra_agent import (
        get_service_status,
        list_recent_deployments,
        deploy_to_staging,
        deploy_to_production,
        rollback_deployment,
        run_migration,
        modify_firewall_rule,
        rotate_credentials,
    )

    # Load configuration
    attesta = Attesta.from_config("attesta.yaml")
    llm = ChatOpenAI(model="gpt-4o")

    # Create LangChain tools from the gated functions
    tools = [
        StructuredTool.from_function(get_service_status),
        StructuredTool.from_function(list_recent_deployments),
        StructuredTool.from_function(deploy_to_staging),
        StructuredTool.from_function(deploy_to_production),
        StructuredTool.from_function(rollback_deployment),
        StructuredTool.from_function(run_migration),
        StructuredTool.from_function(modify_firewall_rule),
        StructuredTool.from_function(rotate_credentials),
    ]

    # Wrap tools with Attesta -- risk overrides for known-dangerous ops
    wrapper = AttestaToolWrapper(
        attesta,
        risk_overrides={
            "deploy_to_production": "high",
            "rollback_deployment": "high",
            "run_migration": "critical",
            "modify_firewall_rule": "high",
        },
    )
    protected_tools = wrapper.wrap_tools(tools)

    # Bind tools to the LLM
    llm_with_tools = llm.bind_tools(protected_tools)


    # Define graph nodes
    async def agent_node(state: MessagesState) -> MessagesState:
        response = await llm_with_tools.ainvoke(state["messages"])
        return {"messages": [response]}


    def should_continue(state: MessagesState) -> str:
        last_msg = state["messages"][-1]
        if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
            return "gate"
        return "__end__"


    # Build the LangGraph
    builder = StateGraph(MessagesState)
    builder.add_node("agent", agent_node)
    builder.add_node("gate", attesta_node(attesta))
    builder.add_node("tools", ToolNode(protected_tools))

    builder.set_entry_point("agent")
    builder.add_conditional_edges("agent", should_continue)
    builder.add_edge("gate", "tools")
    builder.add_edge("tools", "agent")

    graph = builder.compile()


    async def run_infra_agent(user_message: str) -> str:
        """Run the infrastructure agent with Attesta gating."""
        result = await graph.ainvoke(
            {"messages": [("user", user_message)]}
        )
        for msg in reversed(result["messages"]):
            if hasattr(msg, "content") and msg.content:
                return msg.content
        return "No response"


    if __name__ == "__main__":
        # Scenario 1: Status check (LOW)
        print("--- Scenario 1: Service Status (LOW) ---")
        result = asyncio.run(
            run_infra_agent(
                "What is the current status of the api-gateway service?"
            )
        )
        print(result)

        # Scenario 2: Staging deployment (MEDIUM)
        print("\n--- Scenario 2: Staging Deploy (MEDIUM) ---")
        result = asyncio.run(
            run_infra_agent(
                "Deploy api-gateway v2.1.0 to staging using a "
                "rolling update strategy."
            )
        )
        print(result)

        # Scenario 3: Production deployment (CRITICAL)
        print("\n--- Scenario 3: Production Deploy (CRITICAL) ---")
        result = asyncio.run(
            run_infra_agent(
                "Deploy api-gateway v2.1.0 to production with a 10% "
                "canary rollout. Change ticket is CHG-2026-0042."
            )
        )
        print(result)

        # Scenario 4: Database migration (CRITICAL)
        print("\n--- Scenario 4: Database Migration (CRITICAL) ---")
        result = asyncio.run(
            run_infra_agent(
                "Run migration 20260213_add_user_preferences.sql on the "
                "primary database in production."
            )
        )
        print(result)

        # Scenario 5: Emergency rollback (HIGH)
        print("\n--- Scenario 5: Emergency Rollback (HIGH) ---")
        result = asyncio.run(
            run_infra_agent(
                "Rollback api-gateway to v2.0.3 in production. "
                "Reason: v2.1.0 is causing 502 errors on /api/checkout."
            )
        )
        print(result)

        # Scenario 6: Firewall change (HIGH)
        print("\n--- Scenario 6: Firewall Rule Change (HIGH) ---")
        result = asyncio.run(
            run_infra_agent(
                "Open port 443 for ingress from CIDR 10.0.0.0/8 on "
                "firewall rule FW-PROD-001. Protocol TCP."
            )
        )
        print(result)
    ```
  </Step>

  <Step title="Run the agent">
    ```bash theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
    export OPENAI_API_KEY="sk-..."
    python infra_loop.py
    ```
  </Step>
</Steps>

***

## TypeScript Implementation

<Steps>
  <Step title="Install dependencies">
    ```bash theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
    npm install @kyberon/attesta @langchain/openai @langchain/langgraph @langchain/core zod
    ```
  </Step>

  <Step title="Build the infrastructure agent">
    ```typescript infra-agent.ts theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
    import { ChatOpenAI } from "@langchain/openai";
    import { StateGraph, MessagesAnnotation } from "@langchain/langgraph";
    import { ToolNode } from "@langchain/langgraph/prebuilt";
    import { tool } from "@langchain/core/tools";
    import { z } from "zod";
    import { Attesta } from "@kyberon/attesta";
    import { gatedTool, createGateNode } from "@kyberon/attesta/integrations";

    // Load configuration (optionally with a custom domain profile)
    const attesta = new Attesta();

    // ─── Define infrastructure tools ───────────────────────────────────

    const getServiceStatus = tool(
      async ({ service, environment }) => {
        return JSON.stringify({
          service,
          environment,
          status: "healthy",
          version: "2.0.3",
          instances: 4,
          uptime_hours: 720,
        });
      },
      {
        name: "get_service_status",
        description: "Check the health and deployment status of a service.",
        schema: z.object({
          service: z.string().describe("Service name"),
          environment: z
            .string()
            .default("production")
            .describe("Target environment"),
        }),
      }
    );

    const deployToProduction = tool(
      async ({ service, version, strategy, canaryPercentage, changeTicket }) => {
        return JSON.stringify({
          deployment_id: "DEP-PROD-2026-042",
          service,
          version,
          environment: "production",
          strategy,
          canary_percentage: canaryPercentage,
          change_ticket: changeTicket,
          status: "canary_deployed",
        });
      },
      {
        name: "deploy_to_production",
        description:
          "Deploy a service to production. Requires SRE and release manager approval.",
        schema: z.object({
          service: z.string(),
          version: z.string(),
          strategy: z
            .enum(["canary", "rolling", "blue-green"])
            .default("canary"),
          canaryPercentage: z.number().default(10),
          changeTicket: z.string().default(""),
        }),
      }
    );

    const rollbackDeployment = tool(
      async ({ service, targetVersion, reason }) => {
        return JSON.stringify({
          rollback_id: "RBK-2026-007",
          service,
          from_version: "2.1.0",
          to_version: targetVersion,
          reason,
          status: "rolled_back",
        });
      },
      {
        name: "rollback_deployment",
        description:
          "Rollback a production service to a previous version. Emergency operation.",
        schema: z.object({
          service: z.string(),
          targetVersion: z.string(),
          reason: z.string(),
        }),
      }
    );

    const runMigration = tool(
      async ({ migrationFile, database, environment }) => {
        return JSON.stringify({
          migration_id: "MIG-2026-015",
          migration_file: migrationFile,
          database,
          environment,
          tables_affected: ["users", "orders"],
          status: "migration_applied",
        });
      },
      {
        name: "run_migration",
        description:
          "Run a database schema migration. Requires DBA and backend lead approval.",
        schema: z.object({
          migrationFile: z.string(),
          database: z.string(),
          environment: z.string().default("production"),
        }),
      }
    );

    const modifyFirewallRule = tool(
      async ({ ruleId, action, protocol, port, cidr, direction }) => {
        return JSON.stringify({
          rule_id: ruleId,
          action,
          protocol,
          port,
          cidr,
          direction,
          status: "rule_updated",
        });
      },
      {
        name: "modify_firewall_rule",
        description:
          "Modify a firewall rule. Affects network security boundary.",
        schema: z.object({
          ruleId: z.string(),
          action: z.enum(["allow", "deny"]),
          protocol: z.enum(["tcp", "udp", "icmp"]),
          port: z.number(),
          cidr: z.string(),
          direction: z.enum(["ingress", "egress"]).default("ingress"),
        }),
      }
    );

    // ─── Gate the tools ────────────────────────────────────────────────

    const gatedTools = [
      gatedTool(getServiceStatus, {
        agentId: "infra-agent",
        environment: "production",
      }),
      gatedTool(deployToProduction, {
        agentId: "infra-agent",
        environment: "production",
        riskHints: { production: true },
      }),
      gatedTool(rollbackDeployment, {
        agentId: "infra-agent",
        environment: "production",
        riskHints: { production: true, emergency: true },
      }),
      gatedTool(runMigration, {
        agentId: "infra-agent",
        environment: "production",
        riskHints: { production: true, affects_database: true },
      }),
      gatedTool(modifyFirewallRule, {
        agentId: "infra-agent",
        environment: "production",
        riskHints: { security_boundary: true },
      }),
    ];

    // ─── Build the LangGraph ───────────────────────────────────────────

    const llm = new ChatOpenAI({ model: "gpt-4o" });
    const llmWithTools = llm.bindTools(gatedTools);

    async function agentNode(state: typeof MessagesAnnotation.State) {
      const response = await llmWithTools.invoke(state.messages);
      return { messages: [response] };
    }

    function shouldContinue(state: typeof MessagesAnnotation.State): string {
      const lastMsg = state.messages.at(-1);
      if (
        lastMsg &&
        "tool_calls" in lastMsg &&
        (lastMsg as any).tool_calls?.length > 0
      ) {
        return "gate";
      }
      return "__end__";
    }

    const gateNode = createGateNode({
      actionName: "infra-operation",
      agentId: "infra-agent",
      environment: "production",
    });

    const graph = new StateGraph(MessagesAnnotation)
      .addNode("agent", agentNode)
      .addNode("gate", gateNode)
      .addNode("tools", new ToolNode(gatedTools))
      .addConditionalEdges("agent", shouldContinue)
      .addEdge("gate", "tools")
      .addEdge("tools", "agent")
      .compile();

    // ─── Run scenarios ─────────────────────────────────────────────────

    async function main() {
      console.log("--- Scenario 1: Service Status (LOW) ---");
      const status = await graph.invoke({
        messages: [
          { role: "user", content: "What is the status of api-gateway?" },
        ],
      });
      console.log(status.messages.at(-1)?.content);

      console.log("\n--- Scenario 2: Production Deploy (CRITICAL) ---");
      const deploy = await graph.invoke({
        messages: [
          {
            role: "user",
            content:
              "Deploy api-gateway v2.1.0 to production with 10% canary. " +
              "Change ticket CHG-2026-0042.",
          },
        ],
      });
      console.log(deploy.messages.at(-1)?.content);

      console.log("\n--- Scenario 3: Database Migration (CRITICAL) ---");
      const migration = await graph.invoke({
        messages: [
          {
            role: "user",
            content:
              "Run migration 20260213_add_user_preferences.sql " +
              "on the primary database in production.",
          },
        ],
      });
      console.log(migration.messages.at(-1)?.content);

      console.log("\n--- Scenario 4: Emergency Rollback (HIGH) ---");
      const rollback = await graph.invoke({
        messages: [
          {
            role: "user",
            content:
              "Rollback api-gateway to v2.0.3. " +
              "v2.1.0 is causing 502 errors on /api/checkout.",
          },
        ],
      });
      console.log(rollback.messages.at(-1)?.content);

      console.log("\n--- Scenario 5: Firewall Change (HIGH) ---");
      const firewall = await graph.invoke({
        messages: [
          {
            role: "user",
            content:
              "Allow TCP ingress on port 443 from 10.0.0.0/8 " +
              "on rule FW-PROD-001.",
          },
        ],
      });
      console.log(firewall.messages.at(-1)?.content);
    }

    main().catch(console.error);
    ```
  </Step>

  <Step title="Run the agent">
    ```bash theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
    export OPENAI_API_KEY="sk-..."
    npx tsx infra-agent.ts
    ```
  </Step>
</Steps>

***

## Expected Terminal Output

### Scenario 1: Service Status (LOW -- auto-approved)

Read-only status check. No deployment or destructive patterns match. Auto-approved silently.

| Field          | Value                |
| -------------- | -------------------- |
| **Risk Level** | LOW                  |
| **Action**     | `get_service_status` |
| **Risk Score** | 0.10                 |
| **Domain**     | custom (SOC 2)       |
| **Result**     | AUTO-APPROVED        |

### Scenario 2: Staging Deployment (MEDIUM -- confirm)

The `deploy` keyword triggers the deployment pattern (contribution: 0.8), but the `staging` environment hint keeps the risk in MEDIUM range. A simple confirmation is required.

| Field          | Value                                                            |
| -------------- | ---------------------------------------------------------------- |
| **Risk Level** | MEDIUM                                                           |
| **Action**     | `deploy_to_staging`                                              |
| **Risk Score** | 0.42                                                             |
| **Domain**     | custom (SOC 2)                                                   |
| **Arguments**  | `service="api-gateway"`, `version="2.1.0"`, `strategy="rolling"` |
| **Challenge**  | CONFIRM (min review: 3s)                                         |
| **Prompt**     | Deploy api-gateway v2.1.0 to staging? \[y/N] **y**               |
| **Result**     | APPROVED (reviewed in 4.2s)                                      |

### Scenario 3: Production Deployment (CRITICAL -- multi-party)

The same `deploy` function name, but with the `production: True` hint, triggers the production multiplier (1.6x). The combined score crosses into CRITICAL, requiring dual approval from the SRE lead and release manager.

| Field          | Value                                                                                                                    |
| -------------- | ------------------------------------------------------------------------------------------------------------------------ |
| **Risk Level** | CRITICAL                                                                                                                 |
| **Action**     | `deploy_to_production`                                                                                                   |
| **Risk Score** | 0.88                                                                                                                     |
| **Domain**     | custom (SOC 2)                                                                                                           |
| **Arguments**  | `service="api-gateway"`, `version="2.1.0"`, `strategy="canary"`, `canary_percentage=10`, `change_ticket="CHG-2026-0042"` |
| **Escalation** | Production deployment                                                                                                    |
| **Notified**   | sre\_lead, release\_manager                                                                                              |
| **Challenge**  | MULTI-PARTY (2 of 2 required)                                                                                            |

**Approver 1** ([sre.lead@company.com](mailto:sre.lead@company.com)) -- TEACH-BACK:

> Deploying api-gateway v2.1.0 to production with a 10% canary rollout under change ticket CHG-2026-0042. This will update 1 of 10 instances initially.
>
> Approved (48.2s)

**Approver 2** ([release.mgr@company.com](mailto:release.mgr@company.com)) -- QUIZ:

> Q1: Which service is being deployed? **api-gateway**
> Q2: What is the change ticket? **CHG-2026-0042**
>
> Approved (35.4s)

**Result:** APPROVED (2/2 approvers, total 83.6s)

### Scenario 4: Database Migration (CRITICAL -- multi-party)

The `run_migration` function is overridden to CRITICAL risk. A custom domain's schema migration escalation rule can require DBA and backend lead approval.

| Field          | Value                                                                                                  |
| -------------- | ------------------------------------------------------------------------------------------------------ |
| **Risk Level** | CRITICAL                                                                                               |
| **Action**     | `run_migration`                                                                                        |
| **Risk Score** | 0.91                                                                                                   |
| **Domain**     | custom (SOC 2)                                                                                         |
| **Arguments**  | `migration_file="20260213_add_user_preferences.sql"`, `database="primary"`, `environment="production"` |
| **Escalation** | Schema migration                                                                                       |
| **Notified**   | dba, backend\_lead                                                                                     |
| **Challenge**  | MULTI-PARTY (2 of 2 required)                                                                          |

**Approver 1** ([dba@company.com](mailto:dba@company.com)) -- TEACH-BACK:

> Running migration 20260213\_add\_user\_preferences.sql on the primary production database. This will alter the users and orders tables to add preference columns.
>
> Approved (55.1s)

**Approver 2** ([backend.lead@company.com](mailto:backend.lead@company.com)) -- QUIZ:

> Q1: Which database is affected? **primary**
> Q2: Is this migration reversible? **No (DDL)**
>
> Approved (41.3s)

**Result:** APPROVED (2/2 approvers, total 96.4s)

### Scenario 5: Emergency Rollback (HIGH -- quiz)

Rollbacks are emergency operations. Despite the urgency, Attesta requires a quiz challenge to verify the operator understands the rollback impact. The 15-second minimum review prevents panic-driven approvals.

| Field          | Value                                                                                              |
| -------------- | -------------------------------------------------------------------------------------------------- |
| **Risk Level** | HIGH                                                                                               |
| **Action**     | `rollback_deployment`                                                                              |
| **Risk Score** | 0.72                                                                                               |
| **Domain**     | custom (SOC 2)                                                                                     |
| **Arguments**  | `service="api-gateway"`, `target_version="2.0.3"`, `reason="v2.1.0 causing 502s on /api/checkout"` |
| **Challenge**  | QUIZ (min review: 15s)                                                                             |

> **Q1:** Which version will the service revert to? **2.0.3** -- Correct
>
> **Q2:** What is the reported issue? **502 errors on the checkout endpoint** -- Correct
>
> **Q3:** How many production instances will be reverted? **All 10** -- Correct

**Result:** APPROVED (reviewed in 23.7s)

<Warning>
  Rollback protection is deliberate. During incidents, the instinct is to approve everything instantly. Attesta's minimum review time (15s for HIGH) prevents this while keeping the delay short enough for genuine emergencies. If you need faster rollbacks, lower the `high` review time -- but never remove it entirely.
</Warning>

### Scenario 6: Firewall Rule Change (HIGH -- teach-back)

Firewall changes trigger the custom domain's firewall modification escalation rule, requiring teach-back verification. The security engineer and network admin are notified.

| Field          | Value                                                                                                               |
| -------------- | ------------------------------------------------------------------------------------------------------------------- |
| **Risk Level** | HIGH                                                                                                                |
| **Action**     | `modify_firewall_rule`                                                                                              |
| **Risk Score** | 0.78                                                                                                                |
| **Domain**     | custom (SOC 2)                                                                                                      |
| **Arguments**  | `rule_id="FW-PROD-001"`, `action="allow"`, `protocol="TCP"`, `port=443`, `cidr="10.0.0.0/8"`, `direction="ingress"` |
| **Escalation** | Firewall modification                                                                                               |
| **Notified**   | security\_engineer, network\_admin                                                                                  |
| **Challenge**  | TEACH-BACK (min review: 15s)                                                                                        |

> **Explain what this firewall change will do:**
> This will allow inbound TCP traffic on port 443 from the 10.0.0.0/8 CIDR range (internal network) through firewall rule FW-PROD-001. This opens HTTPS access from internal services to the production environment.
>
> Key terms matched: allow, TCP, 443, 10.0.0.0/8, ingress

**Result:** APPROVED (reviewed in 34.8s)

***

## Audit Trail

After running the scenarios, the audit log at `.attesta/infra-audit.jsonl` contains entries for every operation:

```json theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
{
  "entry_id": "aud_d3e4f5a6",
  "chain_hash": "e8f2a1b3c5d7...e9f1a3b5c7d9e1",
  "previous_hash": "b4c7d9e1f3a5...c9d1e3f5a7b9c1",
  "action_name": "deploy_to_production",
  "action_description": "Deploy api-gateway v2.1.0 to production (canary 10%)",
  "agent_id": "infra-agent",
  "risk_score": 0.88,
  "risk_level": "critical",
  "challenge_type": "multi_party",
  "challenge_passed": true,
  "approver_ids": [
    "sre.lead@company.com",
    "release.mgr@company.com"
  ],
  "verdict": "approved",
  "review_duration_seconds": 83.6,
  "min_review_met": true,
  "intercepted_at": "2026-02-13T10:30:00.000Z",
  "decided_at": "2026-02-13T10:31:23.600Z",
  "executed_at": "2026-02-13T10:31:23.750Z",
  "session_id": "sess_infra_g7h8i9",
  "environment": "production",
  "metadata": {
    "domain": "my-infra",
    "service": "api-gateway",
    "version": "2.1.0",
    "strategy": "canary",
    "change_ticket": "CHG-2026-0042",
    "soc2_control": "CC8.1"
  }
}
```

Verify and query the infrastructure audit trail:

```bash theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
# Verify chain integrity
attesta audit verify .attesta/infra-audit.jsonl

# Output:
# ✓ Chain integrity verified (6 entries)
# ✓ All minimum review times met
# ✓ Multi-party approvals validated

# Export all production deployments
attesta audit export --action "deploy_to_production" .attesta/infra-audit.jsonl

# Export all denied actions (useful for incident post-mortems)
attesta audit export --verdict denied .attesta/infra-audit.jsonl

# Check for rubber-stamped rollbacks during incidents
attesta audit rubber-stamps .attesta/infra-audit.jsonl
```

***

## Risk Level Summary

| Function                  | Risk Level | Challenge                      | Key Trigger                                   |
| ------------------------- | ---------- | ------------------------------ | --------------------------------------------- |
| `get_service_status`      | LOW        | Auto-approve                   | Read-only status check                        |
| `list_recent_deployments` | LOW        | Auto-approve                   | Read-only history                             |
| `deploy_to_staging`       | MEDIUM     | Confirm (3s)                   | Deploy pattern + staging environment          |
| `deploy_to_production`    | CRITICAL   | Multi-party, 2 approvers (45s) | Deploy pattern + production multiplier (1.6x) |
| `rollback_deployment`     | HIGH       | Quiz (15s)                     | Rollback pattern + emergency hint             |
| `run_migration`           | CRITICAL   | Multi-party, 2 approvers (45s) | Risk override + schema migration escalation   |
| `modify_firewall_rule`    | HIGH       | Teach-back (15s)               | Firewall pattern + security boundary hint     |
| `rotate_credentials`      | HIGH       | Quiz (15s)                     | Credential management pattern                 |

<Tip>
  A custom infrastructure domain profile can assign lower risk to canary and blue-green deployments (contribution: 0.5) compared to full rollouts (contribution: 0.8). If you primarily use canary deployments, you may find that production deploys land in HIGH rather than CRITICAL. Adjust the `production` amplifier boost if you want all production deploys to require multi-party approval regardless of strategy.
</Tip>

***

## Compliance Mapping

| Framework                                   | Attesta Mechanism                                                 |
| ------------------------------------------- | ----------------------------------------------------------------- |
| SOC 2 CC6.1 -- Logical access security      | Access control patterns flag permission and credential changes    |
| SOC 2 CC7.2 -- Monitoring system components | Audit trail records all infrastructure operations                 |
| SOC 2 CC8.1 -- Change management            | Multi-party approval for production deployments and migrations    |
| ISO 27001 A.9 -- Access control             | Credential rotation and permission changes require verification   |
| ISO 27001 A.12 -- Operations security       | Production multiplier escalates risk for live environment changes |
| ISO 27001 A.13 -- Communications security   | Firewall patterns with teach-back challenges                      |

***

## Next Steps

<CardGroup cols={2}>
  <Card title="Custom Domain Profiles" icon="server" href="/domains/custom-domains">
    Build your own domain-specific risk profiles
  </Card>

  <Card title="Multi-Framework" icon="puzzle-piece" href="/examples/multi-framework">
    Same config across LangChain, OpenAI, and Claude
  </Card>

  <Card title="LangChain Integration" icon="link-horizontal" href="/integrations/langchain">
    Full LangChain and LangGraph reference
  </Card>
</CardGroup>