Continuous & Automated TRA

"""
pytm Threat Model: Payment API Service
Generates STRIDE threats and DFD automatically.
"""
from pytm import TM, Server, Datastore, Dataflow, Boundary, Actor

# Define the threat model
tm = TM("Payment API - Threat Model")
tm.description = "TRA for payment processing microservice"
tm.isOrdered = True

# Trust boundaries
internet = Boundary("Internet")
dmz = Boundary("DMZ")
internal = Boundary("Internal Network")
db_zone = Boundary("Database Zone")

# External entities
customer = Actor("Customer")
customer.inBoundary = internet

payment_provider = Actor("Payment Provider (Stripe)")
payment_provider.inBoundary = internet

# Processes
api_gateway = Server("API Gateway")
api_gateway.inBoundary = dmz
api_gateway.protocol = "HTTPS"
api_gateway.isEncrypted = True
api_gateway.sanitizesInput = True

payment_service = Server("Payment Service")
payment_service.inBoundary = internal
payment_service.protocol = "gRPC"
payment_service.isEncrypted = True
payment_service.hasAccessControl = True
payment_service.authorizesSource = True

# Data stores
payment_db = Datastore("Payment Database")
payment_db.inBoundary = db_zone
payment_db.isEncrypted = True
payment_db.isSQL = True
payment_db.hasAccessControl = True

audit_log = Datastore("Audit Log (Immutable)")
audit_log.inBoundary = db_zone
audit_log.isEncrypted = True

# Data flows
customer_to_gw = Dataflow(customer, api_gateway, "Payment Request")
customer_to_gw.protocol = "HTTPS"
customer_to_gw.isEncrypted = True
customer_to_gw.sanitizesInput = True

gw_to_service = Dataflow(api_gateway, payment_service, "Validated Request")
gw_to_service.protocol = "gRPC + mTLS"
gw_to_service.isEncrypted = True

service_to_db = Dataflow(payment_service, payment_db, "Transaction Record")
service_to_db.protocol = "TLS"
service_to_db.isEncrypted = True

service_to_provider = Dataflow(
    payment_service, payment_provider, "Payment Authorization"
)
service_to_provider.protocol = "HTTPS"
service_to_provider.isEncrypted = True

service_to_audit = Dataflow(
    payment_service, audit_log, "Audit Event"
)
service_to_audit.protocol = "TLS"
service_to_audit.isEncrypted = True

# Generate threats and DFD
tm.process()
# Output: DFD diagram + STRIDE threat list + HTML report

# Threagile YAML Model: Payment Microservice
# Run: docker run --rm -v "$(pwd)":/app/work threagile/threagile -model /app/work/model.yaml

threagile_version: 1.0.0

title: Payment Service TRA
date: 2025-01-15

business_criticality: critical

technical_assets:
  api-gateway:
    type: process
    usage: business
    technology: kong
    internet: true
    encryption: transparent
    communication_links:
      payment-service-link:
        target: payment-service
        protocol: grpc
        authentication: certificate
        encryption: transparent

  payment-service:
    type: process
    usage: business
    technology: go
    internet: false
    encryption: transparent
    data_assets_processed:
      - cardholder-data
      - transaction-records
    communication_links:
      db-link:
        target: payment-db
        protocol: postgresql
        authentication: credentials
        encryption: transparent
      stripe-link:
        target: stripe-api
        protocol: https
        authentication: token
        encryption: transparent

  payment-db:
    type: datastore
    usage: business
    technology: postgresql
    internet: false
    encryption: data-with-symmetric-shared-key
    data_assets_stored:
      - cardholder-data
      - transaction-records

data_assets:
  cardholder-data:
    usage: business
    quantity: many
    confidentiality: strictly-confidential
    integrity: critical
    availability: critical

  transaction-records:
    usage: business
    quantity: many
    confidentiality: confidential
    integrity: critical
    availability: critical

trust_boundaries:
  dmz:
    type: network-cloud-security-group
    technical_assets_inside:
      - api-gateway

  internal:
    type: network-cloud-security-group
    technical_assets_inside:
      - payment-service

  data-zone:
    type: network-cloud-security-group
    technical_assets_inside:
      - payment-db

# GitHub Actions Workflow: Automated Threat Model Check
name: Threat Model Validation
on:
  pull_request:
    paths:
      - 'architecture/**'
      - 'threat-model/**'
      - 'infrastructure/**'

jobs:
  threat-model:
    runs-on: ubuntu-latest
    permissions:
      contents: read
      pull-requests: write

    steps:
      - uses: actions/checkout@v4

      - name: Run Threagile Analysis
        run: |
          docker run --rm \
            -v "${{ github.workspace }}/threat-model:/app/work" \
            threagile/threagile \
            -model /app/work/model.yaml \
            -output /app/work/output

      - name: Check for High/Critical Risks
        run: |
          # Parse Threagile JSON output for risk severity
          HIGH_RISKS=$(jq '[.identified_risks[] |
            select(.severity == "high" or
                   .severity == "critical")] |
            length' threat-model/output/risks.json)

          if [ "$HIGH_RISKS" -gt 0 ]; then
            echo "::error::Found $HIGH_RISKS high/critical risks"
            echo "## Threat Model: $HIGH_RISKS Risks Found" >> $GITHUB_STEP_SUMMARY
            jq -r '.identified_risks[] |
              select(.severity == "high" or
                     .severity == "critical") |
              "- **(.severity)**: (.title)"' \
              threat-model/output/risks.json >> $GITHUB_STEP_SUMMARY
            exit 1
          fi

      - name: Upload Threat Report
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: threat-model-report
          path: threat-model/output/

# Risk-as-Code Policy (Rego / Open Policy Agent)
# Evaluates architecture decisions against risk policies

package tra.risk_policies

# Deny unencrypted data stores containing sensitive data
deny[msg] {
    some store in input.data_stores
    store.classification in ["confidential", "restricted"]
    not store.encryption_at_rest
    msg := sprintf(
        "RISK: Data store '%s' contains %s data without encryption at rest",
        [store.name, store.classification]
    )
}

# Deny internet-facing services without authentication
deny[msg] {
    some service in input.services
    service.internet_facing
    not service.authentication_required
    msg := sprintf(
        "RISK: Internet-facing service '%s' has no authentication",
        [service.name]
    )
}

# Warn on single-tenant database without backup encryption
warn[msg] {
    some store in input.data_stores
    store.business_criticality == "critical"
    not store.backup_encryption
    msg := sprintf(
        "WARN: Critical data store '%s' backups are not encrypted",
        [store.name]
    )
}

# Deny services processing PII without audit logging
deny[msg] {
    some service in input.services
    service.processes_pii
    not service.audit_logging_enabled
    msg := sprintf(
        "RISK: Service '%s' processes PII without audit logging",
        [service.name]
    )
}