Choreo ThreatModel: Comprehensive Walkthrough

# For Hex publication/readers:
Mix.install([
  {:choreo, "~> 0.14.1"},
  # {:choreo, path: Path.expand("../..", __DIR__), force: true},
  {:kino_vizjs, "~> 0.9.0"}
])

# For local development:
# Mix.install([
#   {:choreo, path: Path.expand("~/repos/elixir/choreo")},
#   {:kino_vizjs, "~> 0.9.0"}
# ])

Section

Rendering diagrams: This livebook uses Choreo.Lab.Siren and Choreo.Lab.Sketch for Mermaid diagrams, plus Kino.VizJS for DOT/Graphviz views. Some sequence examples also use Livebook's native Kino.Mermaid support.

What is Threat Modeling?

Threat modeling is a structured way to identify, quantify, and address security risks in a system. Instead of waiting for a penetration test to reveal vulnerabilities, you analyze the architecture before writing code and ask: "What could go wrong?"

STRIDE is a popular mnemonic for threat categories:

Choreo.ThreatModel lets you describe your architecture as code, define trust boundaries, and automatically generate STRIDE threats with severity scoring.

API Approaches: Programmatic Pipe API vs Lab DSL

Choreo provides two complementary approaches to threat modeling:

1. Programmatic Pipe API (Choreo.ThreatModel): The canonical, explicit interface (ThreatModel.new() |> ThreatModel.add_trust_boundary(...) |> ThreatModel.add_process(...) |> ThreatModel.data_flow(...)). Ideal for automated security scans, static code ingestion, and CI/CD policy pipelines.
2. Lab DSL (Choreo.Lab.DSL.ThreatModel): An expressive sketch syntax using threat_model do ... end, semantic element constructors (external_entity, process, data_store, service, db), scoped boundary blocks (boundary "Name", level: 0 do ... end), direct flows (~>), and security modifiers (|> encrypted("HTTPS", protocol: :https)). Ideal for Livebooks, architectural risk reviews, and rapid STRIDE sketching.

Example 1 demonstrates the canonical programmatic pipe API, while subsequent examples showcase the concise Lab DSL.

Example 1: Classic Three-Tier Web Application

Let's start with the simplest useful model: a user, a web API, and a database. This mirrors the classic pytm getting-started example.

import Choreo.Lab.DSL.ThreatModel
alias Choreo.ThreatModel
alias Choreo.ThreatModel.Analysis

web_app =
  ThreatModel.new()
  |> ThreatModel.add_trust_boundary("internet", level: 0, label: "Internet")
  |> ThreatModel.add_trust_boundary("app", level: 2, label: "Application Zone")
  |> ThreatModel.add_trust_boundary("data", level: 3, label: "Data Layer")
  |> ThreatModel.add_external_entity(:user,
    label: "Customer",
    boundary: "internet",
    description: "End user browsing the site"
  )
  |> ThreatModel.add_process(:web_api,
    label: "Web API",
    boundary: "app",
    privilege: :user,
    description: "Handles HTTP requests"
  )
  |> ThreatModel.add_data_store(:postgres,
    label: "Postgres",
    boundary: "data",
    sensitivity: :confidential,
    description: "Primary application database"
  )
  |> ThreatModel.data_flow(:user, :web_api,
    label: "HTTPS login",
    encrypted: true
  )
  |> ThreatModel.data_flow(:web_api, :postgres,
    label: "SQL query"
  )

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(ThreatModel.to_mermaid(web_app)),
  Graphviz: Kino.VizJS.render(ThreatModel.to_dot(web_app)),
  Sketch: Choreo.Lab.Sketch.new(ThreatModel.to_mermaid(web_app))
)

Notice the color coding:

• Red dashed edges cross trust boundaries unencrypted.
• Orange edges cross boundaries but are encrypted.
• Green nodes live in higher-trust zones.

Generating STRIDE Threats

threats = Analysis.stride_threats(web_app)

threats
|> Enum.sort_by(& &1.severity, :desc)
|> Enum.each(fn t ->
  target = if is_tuple(t.target), do: "#{elem(t.target, 0)} → #{elem(t.target, 1)}", else: t.target
  IO.puts("[#{String.upcase(to_string(t.severity))}] #{t.id} — #{t.category} @ #{target}")
  IO.puts("  #{t.description}")
  IO.puts("  Mitigation: #{t.mitigation}\n")
end)

The model automatically flagged the unencrypted web_api → postgres flow as high-risk information disclosure because it crosses from the application zone into the data layer without encryption.

Validation

Analysis.validate(web_app)
|> Enum.each(fn {sev, msg} ->
  icon = if sev == :error, do: "❌", else: "⚠️"
  IO.puts("#{icon} #{msg}")
end)

Example 2: Microservices with an API Gateway

A more realistic architecture: an API Gateway fronts multiple services. Some flows are internal; some cross from the public internet into the VPC.

microservices =
  threat_model do
    boundary "internet", level: 0 do
      mobile_app = external_entity("Mobile App")
      spa = external_entity("Web SPA")
    end

    boundary "dmz", level: 1 do
      api_gateway =
        process("API Gateway",
          privilege: :none,
          description: "Rate limiting, routing, WAF"
        )
    end

    boundary "vpc", level: 2 do
      auth_service =
        process("Auth Service",
          privilege: :admin,
          description: "Issues and validates JWTs"
        )

      order_service = process("Order Service", privilege: :user)

      payment_webhook =
        process("Payment Webhook",
          privilege: :none,
          description: "Receives async callbacks from Stripe"
        )

      orders_db = data_store("Orders DB", sensitivity: :confidential)
      redis_cache = data_store("Redis", sensitivity: :internal)
    end

    mobile_app ~> api_gateway |> encrypted("REST + mTLS")
    spa ~> api_gateway |> encrypted("HTTPS")
    api_gateway ~> auth_service |> encrypted("gRPC")
    api_gateway ~> order_service |> encrypted("gRPC")
    order_service ~> orders_db |> encrypted("SQL/TLS")
    order_service ~> redis_cache |> flow("Redis protocol")
    payment_webhook ~> order_service |> flow("Event bus")
  end

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(ThreatModel.to_mermaid(microservices)),
  Graphviz: Kino.VizJS.render(ThreatModel.to_dot(microservices)),
  Sketch: Choreo.Lab.Sketch.new(ThreatModel.to_mermaid(microservices))
)

Cross-Boundary Analysis

Which flows cross a trust boundary without encryption?

Analysis.unencrypted_boundary_flows(microservices)
|> Enum.each(fn {from, to} ->
  IO.puts("🔓 Unencrypted boundary crossing: #{from} → #{to}")
end)

Exposed Data Stores

Which databases are reachable from an external entity?

Analysis.exposed_data_stores(microservices)
|> IO.inspect(label: "Exposed data stores")

High-Risk Processes

Which processes touch sensitive data?

Analysis.high_risk_processes(microservices)
|> IO.inspect(label: "High-risk processes")

Attack Paths

What are the complete paths from an external entry point to a data store?

Analysis.attack_paths(microservices)
|> Enum.each(fn path ->
  IO.puts("🎯 #{Enum.join(path, " → ")}")
end)

Threat Summary

summary = Analysis.threat_summary(microservices)

IO.puts("Total threats: #{summary.total}")
IO.puts("\nBy severity:")
Enum.each(summary.by_severity, fn {sev, count} ->
  IO.puts("  #{sev}: #{count}")
end)

Example 3: Multi-Tenant SaaS with Custom Compliance Rules

Real organizations have compliance requirements beyond STRIDE. The :rules option lets you inject custom threat generators via the Choreo.ThreatModel.Analysis.Rule behaviour.

defmodule GDPRRule do
  @behaviour Analysis.Rule

  @impl true
  def threats_for_element(_model, id, data) do
    sensitivity = data[:sensitivity]

    if sensitivity in [:confidential, :restricted] do
      [
        %{
          id: "GDPR-#{id}-1",
          category: :compliance,
          target: id,
          description: "#{data.label} stores personal data without documented retention policy.",
          severity: :high,
          mitigation: "Implement automated data purging and Right-to-be-Forgotten endpoints."
        }
      ]
    else
      []
    end
  end

  @impl true
  def threats_for_flow(_model, from, to, meta) do
    # Cross-boundary flows with personal data that are unencrypted need DPA review
    if meta[:encrypted] do
      []
    else
      [
        %{
          id: "GDPR-FLOW-#{from}-#{to}",
          category: :compliance,
          target: {from, to},
          description: "Unencrypted data flow #{from} → #{to} may require Data Processing Agreement.",
          severity: :high,
          mitigation: "Review DPA coverage and SCCs for third-party processors."
        }
      ]
    end
  end
end

saas =
  threat_model do
    boundary "public", level: 0 do
      tenant_admin = external_entity("Tenant Admin")
    end

    boundary "tenant_isolation", level: 2 do
      app_server = process("App Server", privilege: :user)
      tenant_db = data_store("Tenant DB", sensitivity: :confidential, retention: "90d")
    end

    tenant_admin ~> app_server |> encrypted()
    app_server ~> tenant_db |> encrypted()
  end

# Base STRIDE only
base_threats = Analysis.stride_threats(saas)
IO.puts("Base STRIDE threats: #{length(base_threats)}")

# STRIDE + GDPR custom rules
all_threats = Analysis.stride_threats(saas, rules: [GDPRRule])
IO.puts("With GDPR rules: #{length(all_threats)}")

gdpr_only = Enum.filter(all_threats, &(&1.category == :compliance))
IO.puts("GDPR-specific threats:")
Enum.each(gdpr_only, fn t ->
  target = if is_tuple(t.target), do: "flow", else: t.target
  IO.puts("  #{t.id} @ #{target} — #{t.description}")
end)

Kino.VizJS.render(ThreatModel.to_dot(saas))

Example 4: Sequence Diagrams

While Data Flow Diagrams illustrate security boundaries, Sequence Diagrams visualize interactions chronologically. Choreo supports both Mermaid.js and PlantUML formats.

bt = String.duplicate("`", 3)
plantuml_block = bt <> "plantuml\n" <> ThreatModel.to_plantuml(microservices) <> "\n" <> bt

tabs = [
  {"Mermaid", Kino.Mermaid.new(ThreatModel.to_sequence(microservices))},
  {"PlantUML", Kino.Markdown.new(plantuml_block)}
]

Kino.Layout.tabs(tabs)

You can copy these payloads directly into PlantText, Mermaid Live Editor, or your Confluence macro blocks securely.

Deep Dive: Understanding Severity Scoring

Choreo assigns severity based on element properties and flow context. You don't have to guess — the scoring is derived from the model itself.

Severity rules of thumb:

# Experiment: change sensitivity and watch severity shift
restricted_model =
  threat_model do
    boundary "app", level: 2 do
      api = process("API", id: :api, privilege: :admin)
      vault = data_store("Vault", id: :vault, sensitivity: :restricted)
    end

    api ~> vault
  end

Analysis.stride_threats(restricted_model)
|> Enum.filter(&(&1.target == :vault))
|> Enum.each(fn t ->
  IO.puts("#{t.category} → #{t.severity}")
end)

Fixing Issues: Validation-Driven Hardening

Start with a sloppy model, then use validate/1 as a checklist to harden it.

sloppy =
  threat_model do
    api = process("API")
    db = data_store("DB")

    api ~> db
  end

IO.puts("Issues found:")

Analysis.validate(sloppy) |> Enum.each(fn {sev, msg} -> IO.puts(" <#{sev}> #{msg}") end)

Now fix each issue step by step:

hardened =
  threat_model do
    boundary "app", level: 2 do
      api = process("API", privilege: :user)
      db = data_store("DB", sensitivity: :confidential)
    end

    api ~> db |> encrypted()
  end

IO.puts("After hardening:")
Analysis.validate(hardened) |> IO.inspect()

Kino.VizJS.render(ThreatModel.to_dot(hardened))

Rendering for Stakeholders

Dark theme for presentations

Kino.VizJS.render(ThreatModel.to_dot(microservices, theme: :dark))

Example 5: Multigraph (Parallel Data Flows)

Sometimes multiple distinct data payloads cross boundaries between the same entities (e.g., an HTTPS request and a WebSocket notification channel).

parallel_flows =
  threat_model do
    boundary "internet", level: 0 do
      user = external_entity("User")
    end

    boundary "app", level: 2 do
      api = process("API")
    end

    user ~> api |> encrypted("Payload (HTTPS)")
    user ~> api |> unencrypted("Signal (WS)")
  end

IO.puts("Number of data flows: #{length(ThreatModel.flows(parallel_flows))}")
Kino.VizJS.render(ThreatModel.to_dot(parallel_flows))

Example 6: Ingress, Blast Radius & Attack Path Highlighting

Threat modeling goes beyond cataloging static elements: security engineers need to trace ingress vectors, compute worst-case downstream compromise blast radius, and highlight multi-hop attack vectors on diagrams.

review_model =
  threat_model do
    internet = boundary("Internet", level: 0)
    dmz = boundary("DMZ", level: 1)
    internal = boundary("Internal Zone", level: 2)
    vault = boundary("Data Vault", level: 3)

    user = external("End User", boundary: internet, role: :user)
    partner = external("Partner API", boundary: internet, role: :partner)
    proxy = process("Reverse Proxy", boundary: dmz, privilege: :none, controls: [:rate_limiting, :waf])
    api = process("Core API", boundary: internal, privilege: :user, controls: [:auth, :input_validation])
    billing = process("Billing Service", boundary: internal, privilege: :admin)
    db = database("Customer DB", boundary: vault, sensitivity: :restricted, controls: [:encryption_at_rest])
    webhook = external("Partner Webhook", boundary: internet)

    user ~> proxy |> authenticated("Login") |> encrypted()
    partner ~> proxy |> authenticated("Partner API Call") |> encrypted()
    proxy ~> api |> encrypted("Forward request")
    api ~> billing |> carries(:payment_intent, :confidential) |> encrypted()
    billing ~> db |> carries(:card_token, :restricted) |> protects([:tls, :integrity])
    billing ~> webhook |> encrypted("Payment callback")
  end

IO.puts("=== Ingress Entry Points ===")
ThreatModel.entry_points(review_model) |> IO.inspect()

IO.puts("\n=== Egress Exit Points ===")
ThreatModel.exit_points(review_model) |> IO.inspect()

IO.puts("\n=== Blast Radius if Proxy is compromised ===")
ThreatModel.blast_radius(review_model, :proxy) |> IO.inspect()

Visualizing Highlighted Attack Paths

You can highlight the attack paths directly on the diagram using ThreatModel.highlight_attack_paths/2:

highlighted_review = ThreatModel.highlight_attack_paths(review_model)

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(ThreatModel.to_mermaid(highlighted_review)),
  Graphviz: Kino.VizJS.render(ThreatModel.to_dot(highlighted_review)),
  Sketch: Choreo.Lab.Sketch.new(ThreatModel.to_mermaid(highlighted_review))
)

Example 7: Mitigations, Controls & Markdown Threat Matrices

Choreo tracks applied security controls (like MFA, rate limiting, encryption at rest, WAF) and automatically evaluates whether specific STRIDE threats are mitigated.

all_threats = Analysis.stride_threats(review_model)
unmitigated = ThreatModel.unmitigated_threats(review_model)

IO.puts("Total STRIDE Threats: #{length(all_threats)}")
IO.puts("Unmitigated Threats: #{length(unmitigated)}")
IO.puts("Mitigated Threats: #{length(all_threats) - length(unmitigated)}")

# Generate an executive GitHub Flavored Markdown threat table
table_markdown = ThreatModel.to_markdown(review_model)
Kino.Markdown.new(table_markdown)

Example 8: Reviewer Layer — Residual Risk, Control Gaps, and Findings

After the architecture is modeled, reviewers often need a concise checklist: what risk remains, where controls are missing, where sensitive data can leave the system, and which findings should be handled first.

risk_snapshot = %{
  inherent: ThreatModel.Analysis.risk_score(review_model),
  residual: ThreatModel.residual_risk_score(review_model)
}

Kino.Markdown.new("```elixir\n#{inspect(risk_snapshot, pretty: true)}\n```")

Boundary matrices summarize flows between trust zones, including encryption/authentication coverage and maximum observed sensitivity.

review_model
|> ThreatModel.boundary_matrix()
|> inspect(pretty: true)
|> then(fn code ->
  Kino.Markdown.new("```elixir\n#{code}\n```")
end)

Control gaps turn common review heuristics into an actionable checklist.

control_gaps_md =
  review_model
  |> ThreatModel.control_gaps()
  |> Enum.map(fn gap ->
    "* `#{inspect(gap.target)}` — missing #{Enum.map_join(gap.missing, ", ", &to_string/1)}: #{gap.reason}"
  end)
  |> Enum.join("\n")

Kino.Markdown.new(control_gaps_md)

Exfiltration paths reverse the usual attack-path question: "Can confidential or restricted data flow outward to an external party?"

review_model
|> ThreatModel.exfiltration_paths()
|> Enum.map(&Enum.join(&1, " → "))
|> inspect(pretty: true)
|> Kino.Text.new()

For a release review, prioritized_findings/2 combines validation issues, exposed stores, unencrypted boundary flows, high-risk processes, exfiltration paths, and control gaps into a ranked finding list.

findings_md =
  review_model
  |> ThreatModel.prioritized_findings(max_findings: 8)
  |> Enum.map(fn finding ->
    """
    ### #{finding.id}: #{finding.title}

    - Severity: `#{finding.severity}`
    - Kind: `#{finding.kind}`
    - Target: `#{inspect(finding.target)}`
    - Recommendation: #{finding.recommendation}
    """
  end)
  |> Enum.join("\n")

Kino.Markdown.new(findings_md)

Cheat Sheet

Lab DSL Syntax (Choreo.Lab.DSL.ThreatModel)

Programmatic Pipe API & Analysis (Choreo.ThreatModel)

Summary

Threat modeling as code means your security review is version-controlled, diffable, and repeatable. Every pull request can include an updated threat model alongside the code changes.