Threat pattern library for threat modeling

# Threat pattern catalog: extend it with your own patterns and findings.
# applies_to: element types (process, store, external, flow); when: tags a DFD item must carry.
- id: TP-01
  name: Caller-supplied identity
  stride: Spoofing
  applies_to: [flow]
  when: [crosses_boundary]
  check: Is the caller's identity taken from a verified credential, never from a header, field or parameter the caller controls?
  cwe: [CWE-290, CWE-807]
- id: TP-05
  name: Object access by client-supplied key
  stride: Elevation of privilege
  applies_to: [process]
  when: [reads_by_id]
  check: Does every read and write of an object check the caller's right to that object, including exports and bulk routes?
  cwe: [CWE-639]
- id: TP-06
  name: Missing tenant filter in a shared store
  stride: Information disclosure
  applies_to: [store]
  when: [multi_tenant]
  check: Is the tenant condition applied inside every query, cache key and index lookup on this store?
  cwe: [CWE-863]
- id: TP-09
  name: Interpreter injection at the boundary
  stride: Tampering
  applies_to: [process]
  when: [builds_query]
  check: Is untrusted data passed as parameters rather than spliced into SQL, shell, template or query text?
  cwe: [CWE-89, CWE-78, CWE-1336]
  capec: [CAPEC-66, CAPEC-88]
- id: TP-13
  name: Check-then-act race
  stride: Tampering
  applies_to: [process]
  when: [limited_resource]
  check: Are the check and the state change one atomic operation (conditional update, lock or unique constraint)?
  cwe: [CWE-367]
  capec: [CAPEC-29]
- id: TP-14
  name: Replayable message
  stride: Spoofing
  applies_to: [flow]
  when: [inbound_callback]
  check: Is each message signed, time-bounded and checked against already-seen IDs, and is the handler idempotent?
  cwe: [CWE-294, CWE-345]
- id: TP-16
  name: Server-side request to a user-chosen URL
  stride: Elevation of privilege
  applies_to: [process]
  when: [fetches_url]
  check: Are destinations allowlisted or private ranges blocked at connect time, with redirects off and an egress proxy?
  cwe: [CWE-918]
  capec: [CAPEC-664]
- id: TP-20
  name: Sensitive data in logs
  stride: Information disclosure
  applies_to: [store]
  when: [logs]
  check: Are secrets, tokens and personal data kept out of logs, and are logged values encoded against log injection?
  cwe: [CWE-532, CWE-117]
- id: TP-24
  name: Untrusted code with release credentials
  stride: Elevation of privilege
  applies_to: [process]
  when: [ci_pipeline]
  check: Can code from a fork or pull request run in a job that holds publish or deploy credentials?
  cwe: [CWE-829]
- id: TP-27
  name: Unbounded work per request
  stride: Denial of service
  applies_to: [process]
  when: [batch_or_upload]
  check: Are size, count, depth, time and cost limits enforced before the work starts?
  cwe: [CWE-770]
  capec: [CAPEC-130]
- id: TP-29
  name: Untrusted content steering a model's tools
  stride: Elevation of privilege
  applies_to: [process]
  when: [llm_with_tools]
  check: Does a gateway outside the model authorize every tool call against the user's permissions and intent?
  cwe: [CWE-1427]
- id: TP-32
  name: Retrieval without per-user filtering
  stride: Information disclosure
  applies_to: [store]
  when: [vector_index]
  check: Are tenant and permission filters applied inside the vector query, before ranking?
  cwe: [CWE-863]

"""List the threat patterns to review for each element and flow of a data-flow diagram.

Usage: python review_patterns.py patterns.yaml dfd.yaml
"""
import sys

import yaml


def matches(pattern, kind, tags):
    return kind in pattern["applies_to"] and set(pattern["when"]) <= set(tags)


def review(patterns, dfd):
    items = [(e["id"], e["name"], e["type"], e.get("tags", [])) for e in dfd["elements"]]
    items += [(f["id"], f"{f['from']} -> {f['to']}", "flow", f.get("tags", [])) for f in dfd["flows"]]
    for item_id, name, kind, tags in items:
        hits = [p for p in patterns if matches(p, kind, tags)]
        for p in hits:
            refs = ", ".join(p.get("cwe", []) + p.get("capec", []))
            print(f"{item_id:9} {name:22} {p['id']} {p['name']} ({p['stride']}; {refs})")
            print(f"{'':32} {p['check']}")
        if not hits and kind != "external":
            print(f"{item_id:9} {name:22} no catalog pattern: review with STRIDE per element")


if __name__ == "__main__":
    patterns = yaml.safe_load(open(sys.argv[1], encoding="utf-8"))
    dfd = yaml.safe_load(open(sys.argv[2], encoding="utf-8"))
    review(patterns, dfd)