more complicated threat model example

sd.py

#!/usr/bin/env python3
from pytm.pytm import Actor, Boundary, Dataflow, Datastore, Element, ExternalEntity, Process, TM, Server

tm = TM("SecureDrop")
tm.description = "SecureDrop core threat model"

# Trust boundaries
source_area = Boundary("Source Area")
securedrop_area = Boundary("SecureDrop Area")
external_services = Boundary("External Services")
admin_workstation = Boundary("Admin Workstation (Tails)")
publishing_area = Boundary("Publishing Area")
journalist_area = Boundary("Journalist Workstation (Tails)")
airgapped_area = Boundary("Airgapped Viewing Area")

# Users
source = Actor("source")
source.inBoundary = source_area

journalist = Actor("journalist")
# journalist.inBoundary = journalist_area

admin = Actor("administrator")
admin.inBoundary = admin_workstation
admin.isAdmin = True

tor_browser_source = Process("Tor Browser to Source Interface")
tor_browser_source.inBoundary = source_area

tor_browser_journalist = Process("Tor Browser to Journalist Interface")
tor_browser_journalist.inBoundary = journalist_area

ssh_admin = Process("SSH Access to Servers")
ssh_admin.inBoundary = admin_workstation

# TODO: further decompose these areas
app_server = Server("app server")
app_server.inBoundary = securedrop_area
mon_server = Server("mon server")
mon_server.inBoundary = securedrop_area

# Airgap
secure_viewing_station = Boundary("SVS")
secure_viewing_station.inBoundary = airgapped_area

encrypted_document = Element("encrypted submission")
encrypted_document.inBoundary = secure_viewing_station

transfer_device = Datastore("Data Transfer Device")
export_device = Datastore("Export Device")

printer = Datastore("Printer")
printer.inBoundary = airgapped_area

sanitization = Process("Sanitize submission data (MAT)")
sanitization.inBoundary = secure_viewing_station

sanitized_document = Element("Sanitized submission")
sanitized_document.inBoundary = airgapped_area

# Nit for DFD in core: Tor apt repository is still listed in the External Services area
fpf_apt_repository = ExternalEntity("FPF apt server")
ubuntu_apt_repository = ExternalEntity("Ubuntu apt server")
ntp_server = ExternalEntity("NTP server")
smtp_relay = ExternalEntity("SMTP relay")
for external_service in [fpf_apt_repository, ubuntu_apt_repository,
                         ntp_server, smtp_relay]:
    external_service.inBoundary = external_services

# Basic data flows
user_interface_source = Dataflow(source, tor_browser_source,
                                 "tor onion service")
source_to_securedrop = Dataflow(tor_browser_source, app_server,
                                "source tor traffic to app server")
replies_to_source = Dataflow(app_server, tor_browser_source,
                                "journalist replies to source")
admin_to_securedrop_app = Dataflow(ssh_admin, app_server, "ssh")
admin_to_securedrop_mon = Dataflow(ssh_admin, mon_server, "ssh")
user_interface_journalist = Dataflow(journalist, tor_browser_journalist,
                                 "tor authenticated onion service")
journalist_to_securedrop = Dataflow(app_server, tor_browser_journalist,
                                    "journalist downloads documents")
journalist_replies = Dataflow(tor_browser_journalist, app_server,
                              "journalist replies to source")
traverse_airgap_online_side = Dataflow(journalist, transfer_device, "sneakernet")
traverse_airgap_offline_side = Dataflow(transfer_device, encrypted_document, "sneakernet")
export_method_usb = Dataflow(sanitized_document, export_device, "Export to USB")
export_method_printer = Dataflow(sanitized_document, printer, "Export via printing")

# External dataflows
external_fpf_apt_app = Dataflow(fpf_apt_repository, app_server, "FPF apt updates")
external_fpf_apt_mon = Dataflow(fpf_apt_repository, mon_server, "FPF apt updates")
external_ubuntu_apt_app = Dataflow(ubuntu_apt_repository, app_server, "Ubuntu apt updates")
external_ubuntu_apt_mon = Dataflow(ubuntu_apt_repository, mon_server, "Ubuntu apt updates")
external_ntp_app = Dataflow(ntp_server, app_server, "NTP")
external_ntp_mon = Dataflow(ntp_server, mon_server, "NTP")
external_smtp_app = Dataflow(mon_server, smtp_relay, "SMTP")

# Journalist corporate area
corp_workstation = Server("Journalist Corporate Workstation")
corp_workstation.inBoundary = publishing_area
transfer_to_corp_ws = Dataflow(export_device, corp_workstation, "Export to USB")
journalist_processing_exports = Dataflow(journalist, corp_workstation, "Processes submission")

corp_cms = Server("Journalist publishing system")
corp_cms.inBoundary = publishing_area

journalist_sanitizes = Dataflow(sanitization, sanitized_document, "Sanitizes documents")
journalist_uploads_cms = Dataflow(corp_workstation, corp_cms, "Uploads to internal systems")

tm.process()