-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathmain.py
More file actions
6820 lines (6371 loc) · 342 KB
/
Copy pathmain.py
File metadata and controls
6820 lines (6371 loc) · 342 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
import json
import os
from collections import Counter
from contextlib import contextmanager
from datetime import datetime
from io import BytesIO
import plotly.graph_objects as go
import plotly.io as pio
import streamlit as st
from rules_engine import evaluate_threats, compute_risk_score
# ---------------------------------------------------------------------------
# Autosave — persist form answers across Streamlit server restarts
# ---------------------------------------------------------------------------
_ATM_AUTOSAVE = os.path.join(
os.path.dirname(os.path.abspath(__file__)), ".atm_autosave.json"
)
def _autosave():
"""Write all current form answers to disk (called every render cycle).
Navigation state (step_index, report_requested) is intentionally NOT
saved — the app always opens on page 1 so the user fills in fresh."""
try:
payload = {k: st.session_state.get(k) for k in initial_values}
with open(_ATM_AUTOSAVE, "w") as _fh:
json.dump(payload, _fh, default=str)
except Exception:
pass
def _autoload():
"""Restore form answers saved by a previous server run.
Type-validates each value against its default: if the default is a string
but the saved value is an integer (stale Streamlit widget index), the saved
value is discarded and the default is used instead.
"""
try:
if os.path.isfile(_ATM_AUTOSAVE):
with open(_ATM_AUTOSAVE) as _fh:
raw = json.load(_fh)
if not isinstance(raw, dict):
return {}
cleaned = {}
for k, v in raw.items():
default = initial_values.get(k)
# If the default is a string but we loaded an int/float,
# the file stored a widget index — discard it.
if isinstance(default, str) and not isinstance(v, str):
continue
# If the default is a list but we loaded something else, skip.
if isinstance(default, list) and not isinstance(v, list):
continue
cleaned[k] = v
return cleaned
except Exception:
pass
return {}
def _autosave_clear():
"""Remove the autosave file (called when user resets all answers)."""
try:
if os.path.isfile(_ATM_AUTOSAVE):
os.remove(_ATM_AUTOSAVE)
except Exception:
pass
# ---------------------------------------------------------------------------
# Default values for every input key the rule engine reads.
# Keep this in sync with rules_engine.evaluate_threats().
# ---------------------------------------------------------------------------
initial_values = {
# Persona for UX (not consumed by rules engine, filtered out before evaluate_threats)
"persona": "Security Engineer",
# Project context
"project_name": "",
"description": "",
"ai_type": "Traditional ML",
"project_stage": "Development",
"business_impact": "Moderate",
"regulated_domain": [],
"model_type": "Classifier",
"model_source": "Open-source",
"model_updates": "No",
"model_supply_chain_controls": [],
# Fine-tune specific
"fine_tune_data_review": "Not Applicable",
"fine_tune_base_model_trust": "Not Applicable",
# Data & behavior
"training_data": [],
"data_sensitivity": [],
"data_governance_controls": [],
"unlearning_capability": "No",
"external_sources": "No",
"real_time": "No",
"outputs": [],
"output_destinations": [],
"users": [],
"direct_query": "No",
"auto_action": "No",
"plugin_access": "No",
"external_systems": "No",
"can_override": "No",
"user_influence": "No",
# RAG / retrieval / memory
"rag_usage": "No",
"rag_data_sources": [],
"retrieval_access_control": "Not Applicable",
"retrieval_content_filtering": "Not Applicable",
"vector_db_isolation": "Not Applicable",
"embedding_model_provenance": "Not Applicable",
"embedding_inversion_controls": "Not Applicable",
# Deployment
"infra": "AWS",
"exposure": "Public",
"access_control": "None",
"rate_limiting": "No",
"tenant_isolation": "Not Applicable",
"abuse_monitoring": "No",
"cost_monitoring": "No",
"ai_gateway": "No",
"waf": "No",
"network_segmentation": "No",
"env_patching_scanning": "No",
"ci_cd_security": "No",
"backup_rollback": "No",
"incident_response": "No",
# Security controls
"input_validation": "No",
"output_filtering": "No",
"prompt_template": "Not Applicable",
"llm_firewall": [],
"system_prompt_secrets": "Not Applicable",
"system_prompt_protection": "Not Applicable",
"hallucination_controls": "Not Applicable",
"multimodal_injection_testing": "Not Applicable",
"sandboxing": "No",
"data_encrypted_at_rest": "No",
"artifacts_encrypted_at_rest": "No",
"secrets_managed_securely": "No",
"logging": "No",
"auditing": "No",
"adversarial_test_types": [],
"adversarial_test_tools": [],
"red_team": "No",
"safety_evals": "No",
# Governance
"model_card": "No",
"rl_feedback": "No",
"output_watermarking": "No",
"explainability": "No",
# Agentic
"agentic_autonomous": "Not Applicable",
"agentic_tool_access": "Not Applicable",
"agentic_logging": "Not Applicable",
"agentic_hitl": "Not Applicable",
"agentic_sensitive_data": "Not Applicable",
"agentic_malicious_input_detection": "Not Applicable",
"agentic_memory": "Not Applicable",
"agentic_memory_controls": "Not Applicable",
"agentic_identity_scoped": "Not Applicable",
"agentic_code_execution": "Not Applicable",
"agentic_code_sandbox": "Not Applicable",
"agentic_supply_chain_controls": "Not Applicable",
"agentic_kill_switch": "Not Applicable",
"agentic_multi_agent": "Not Applicable",
"agentic_inter_agent_auth": "Not Applicable",
"agentic_plan_inspection": "Not Applicable",
"browser_agent_use": "No",
# MCP
"mcp_usage": "Not Applicable",
"mcp_third_party_servers": "Not Applicable",
"mcp_remote_servers": "Not Applicable",
"mcp_tool_schema_integrity": "Not Applicable",
"mcp_tool_output_sanitization": "Not Applicable",
"mcp_authz": "Not Applicable",
"mcp_human_approval": "Not Applicable",
"mcp_server_isolation": "Not Applicable",
# ---------------------------------------------------------------
# 2026 ADVANCED COVERAGE (added April 2026)
# ---------------------------------------------------------------
# Wiring fix: rules_engine.py reads `data_validation` in 5+ rules
# (data poisoning, RLHF, fine-tune, training-data integrity).
# Previously phantom-read; now wired to a real question.
"data_validation": "No",
# Multi-modal injection (broader than just one yes/no)
"vision_injection_testing": "Not Applicable", # adversarial text in images
"audio_injection_testing": "Not Applicable", # acoustic adversarial / Whisper attacks
"document_pdf_injection": "Not Applicable", # PDF/doc prompt-injection sanitization
"deepfake_detection": "No", # impersonation / synthetic-voice defenses
"c2pa_provenance": "Not Applicable", # C2PA digital content credentials
"watermark_robustness": "Not Applicable", # watermark survives transformations
# Generative-AI specific
"training_data_ip_review": "No", # copyright/IP risk of memorized training data
"model_collapse_monitoring":"Not Applicable", # monitoring for AI-on-AI training collapse
# Fine-tune / customization risks
"lora_adapter_validation": "Not Applicable", # PEFT/LoRA adapter source trust
"alignment_regression_testing": "Not Applicable", # safety eval re-run after fine-tune
"jailbreak_transfer_testing":"Not Applicable", # jailbreak transferability across models
# Inference-time / shared infrastructure
"kv_cache_isolation": "Not Applicable", # KV-cache leakage in multi-tenant inference
# Agentic AI 2026 additions
"agent_collusion_controls": "Not Applicable", # multi-agent infection chain detection
"agent_identity_verification": "Not Applicable", # cryptographic agent identity (vs. just inter-agent auth)
"agent_goal_drift_monitoring": "Not Applicable", # long-horizon goal drift detection
"agent_credential_acquisition": "Not Applicable", # controls on agents requesting new privileges
"agent_hitl_bypass_detection": "Not Applicable", # detect agents trying to bypass HITL gates
# MCP 2026 additions
"mcp_federation_trust": "Not Applicable", # transitive trust in chained MCP servers
"mcp_tool_description_validation":"Not Applicable",# validate JSON tool descriptions
"mcp_prompt_in_result_filtering":"Not Applicable", # sanitize tool RESULTS for embedded prompts
"mcp_transport_security": "Not Applicable", # TLS / signed transport (HTTP/SSE/stdio)
"mcp_shadow_discovery": "Not Applicable", # detect unauthorized MCP server registration
"mcp_audit_telemetry": "Not Applicable", # full MCP invocation logging
# Governance, regulatory & supply-chain transparency
"ai_bom": "No", # AI Bill of Materials maintained
"ai_incident_response_plan":"No", # AI-specific (vs generic) incident plan
"eu_ai_act_high_risk": "Unknown", # is this an EU AI Act high-risk system?
"iso_42001_alignment": "No", # ISO/IEC 42001 AI management alignment
"third_party_audit_ready": "No", # ready for external AI audits
# Classical ML specific
"adversarial_robustness_testing": "No",
"membership_inference_controls": "No",
"model_inversion_controls": "No",
"feature_attribution_available": "No",
"ml_drift_monitoring": "No",
"ml_input_schema_validation": "No",
# Agentic — deployment-level controls (injected into Deployment step)
"agent_action_rate_limit": "No", # per-action cap, separate from API rate limiting
"agent_per_run_budget": "No", # max steps / tokens / cost per single agent run
"agent_execution_isolation": "No", # agent runs in isolated infra vs. shared prod
# Agentic — security-level controls (injected into Security step)
"agent_scope_declared": "No", # explicit allowlist of tools/APIs the agent may call
"agent_tool_output_sanitization": "No", # tool results sanitized before LLM re-ingestion
"agent_destructive_action_gate": "No", # send/delete/write actions require explicit approval
# Agentic — data-scope controls (injected into Data & Behavior step)
"agent_data_write_access": "No", # agent can modify/delete external data stores
"agent_exfil_controls": "No", # controls to prevent agent from leaking data to external APIs
# Reinforcement Learning — specific controls
"rl_reward_hacking_tested": "Not Applicable", # reward function audited / red-teamed
"rl_policy_drift_monitoring": "Not Applicable", # monitoring for policy degradation / env shift
"rl_safe_rl_constraints": "Not Applicable", # safety constraints / shielding in the RL loop
# Edge AI / IoT — specific controls
"edge_ota_update_security": "Not Applicable", # OTA firmware/model update integrity checks
"edge_physical_adversarial": "Not Applicable", # physical adversarial example / patch testing
"edge_model_encryption": "Not Applicable", # model weights encrypted on device
}
# ---------------------------------------------------------------------------
# Pre-built Templates — one-click project bootstrapping
# ---------------------------------------------------------------------------
TEMPLATES = {
"── Select a template ──": {},
"Customer Chatbot (LLM)": {
"project_name": "Customer Support Chatbot",
"ai_type": "Large Language Model (LLM)",
"model_type": "LLM (Generic)",
"model_source": "Third-party API / SaaS",
"project_stage": "Production",
"business_impact": "High",
"exposure": "Public",
"direct_query": "Yes",
"user_influence": "Yes",
"external_sources": "No",
"data_sensitivity": ["PII"],
"users": ["Customers"],
"input_validation": "No",
"output_filtering": "No",
"rate_limiting": "No",
"auditing": "No",
"secrets_managed_securely": "No",
"data_encrypted_at_rest": "No",
},
"RAG Knowledge System": {
"project_name": "Internal RAG Knowledge Assistant",
"ai_type": "RAG / AI Search",
"model_type": "RAG Application",
"model_source": "Pretrained (Vendor)",
"project_stage": "Pilot",
"business_impact": "High",
"exposure": "Authenticated Users Only",
"direct_query": "Yes",
"rag_usage": "Yes",
"external_sources": "Yes",
"data_sensitivity": ["PII", "Credentials/secrets"],
"users": ["Internal employees"],
"input_validation": "No",
"output_filtering": "No",
"retrieval_access_control": "No",
"vector_db_isolation": "No",
"secrets_managed_securely": "No",
},
"Agentic AI Assistant": {
"project_name": "Autonomous AI Agent",
"ai_type": "Agentic AI (e.g., Autonomous Agents)",
"model_type": "Agentic Workflow / Autonomous Agent",
"model_source": "Fine-tuned",
"project_stage": "Development",
"business_impact": "Critical",
"exposure": "Internal-only",
"agentic_autonomous": "Yes",
"agentic_tool_access": "Yes",
"agentic_hitl": "No",
"agentic_memory": "Yes",
"agentic_code_execution": "Yes",
"direct_query": "Yes",
"auto_action": "Yes",
"external_systems": "Yes",
"plugin_access": "Yes",
"input_validation": "No",
"output_filtering": "No",
"secrets_managed_securely": "No",
"users": ["Internal employees"],
},
"ML Fraud Detection": {
"project_name": "ML Fraud Detection Model",
"ai_type": "Traditional ML",
"model_type": "Classifier",
"model_source": "Internally trained from scratch",
"project_stage": "Production",
"business_impact": "Critical",
"exposure": "Back-office batch job",
"data_sensitivity": ["Financial/payment data", "PII"],
"external_sources": "Yes",
"real_time": "Yes",
"users": ["Internal employees"],
"input_validation": "No",
"output_filtering": "No",
"model_card": "No",
"data_encrypted_at_rest": "No",
"secrets_managed_securely": "No",
"auditing": "No",
},
"Generative AI (Image/Media)": {
"project_name": "Generative AI Media System",
"ai_type": "Generative AI (e.g., Image/Audio Generation)",
"model_type": "Multimodal Model",
"model_source": "Open-source",
"project_stage": "Pilot",
"business_impact": "High",
"exposure": "Public",
"direct_query": "Yes",
"user_influence": "Yes",
"external_sources": "No",
"data_sensitivity": [],
"users": ["Customers", "Anonymous"],
"output_filtering": "No",
"rate_limiting": "No",
"auditing": "No",
"model_card": "No",
},
"MCP Tool-Integrated Assistant": {
"project_name": "MCP-Integrated AI Assistant",
"ai_type": "Agentic AI (e.g., Autonomous Agents)",
"model_type": "MCP / Tool-Integrated Assistant",
"model_source": "Third-party API / SaaS",
"project_stage": "Development",
"business_impact": "High",
"exposure": "Internal-only",
"mcp_usage": "Yes",
"mcp_third_party_servers": "Yes",
"mcp_authz": "No",
"mcp_human_approval": "No",
"direct_query": "Yes",
"auto_action": "Yes",
"external_systems": "Yes",
"plugin_access": "Yes",
"secrets_managed_securely": "No",
"auditing": "No",
"users": ["Internal employees"],
},
}
# ---------------------------------------------------------------------------
# Option lists
# ---------------------------------------------------------------------------
AI_TYPE_OPTIONS = [
"Traditional ML",
# Engine has dedicated branches for these classical-ML sub-types — surface
# them in the selector so the specialization isn't dead code. They all
# resolve to the "ml" profile downstream.
"Computer Vision",
"NLP (Non-LLM)",
"Recommendation System",
"Anomaly Detection / Fraud Detection",
"Generative AI (e.g., Image/Audio Generation)",
"Large Language Model (LLM)",
"RAG / AI Search",
"Multimodal AI",
"Agentic AI (e.g., Autonomous Agents)",
]
# Suggested defaults for the model-type field per AI type — keeps the
# questionnaire coherent when a non-default ai_type is selected.
AI_TYPE_DEFAULT_MODEL_TYPE = {
"Traditional ML": "Classifier",
"Computer Vision": "CNN / Computer Vision",
"NLP (Non-LLM)": "Transformer",
"Recommendation System": "Recommender System",
"Anomaly Detection / Fraud Detection":"Clustering / Anomaly Detection",
"Generative AI (e.g., Image/Audio Generation)": "Multimodal Model",
"Large Language Model (LLM)": "LLM (Generic)",
"RAG / AI Search": "RAG Application",
"Multimodal AI": "Multimodal Model",
"Agentic AI (e.g., Autonomous Agents)": "Agentic Workflow / Autonomous Agent",
}
MODEL_TYPE_OPTIONS = [
"Classifier",
"Regression",
"Clustering / Anomaly Detection",
"Recommender System",
"CNN / Computer Vision",
"Transformer",
"Embedding / Vector Search",
"LLM (Generic)",
"LLM (Custom/Fine-tuned)",
"RAG Application",
"Multimodal Model",
"Agentic Workflow / Autonomous Agent",
"Multi-Agent System",
"MCP / Tool-Integrated Assistant",
"Reinforcement Learning System",
"Edge AI / IoT Model",
"Other Custom",
]
MODEL_SOURCE_OPTIONS = [
"Open-source",
"Pretrained (Vendor)",
"Fine-tuned",
"Proprietary",
"Internally trained from scratch",
"Third-party API / SaaS",
"Marketplace model",
"Unknown",
]
PROJECT_STAGE_OPTIONS = ["Idea / PoC", "Development", "Pilot", "Production", "Retired / Decommissioning"]
BUSINESS_IMPACT_OPTIONS = ["Low", "Moderate", "High", "Critical"]
INFRA_OPTIONS = [
"AWS", "GCP", "Azure", "On-prem", "Edge",
"Hybrid / multi-cloud", "SaaS / vendor-hosted", "Developer workstation / local",
]
EXPOSURE_OPTIONS = [
"Public", "Internal-only", "Authenticated Users Only",
"Partner / third-party API", "Back-office batch job",
"Embedded in product", "Developer-only / experimental",
]
ACCESS_CONTROL_OPTIONS = [
"None", "Token-based", "Role-based", "Attribute/policy-based",
"SSO/OIDC", "mTLS/service identity", "Network-only allowlist",
]
# Every control / yes-no question accepts "Not Applicable" so the user can
# explicitly mark a question as irrelevant to their system. The engine scores
# NA as 0 (no gap) and skips firing rules that depend on a missing control.
CONTROL_OPTIONS = ["Yes", "Partial", "No", "Unknown", "Not Applicable"]
YN_OPTIONS = ["Yes", "No", "Not Applicable"]
SEVERITY_ORDER = {"Critical": 0, "High": 1, "Medium": 2, "Low": 3}
# Bug 11: distinct, easy-to-tell-apart colors for severity charts
SEVERITY_COLORS = {
"Critical": "#7f1d1d", # deep maroon
"High": "#ef4444", # bright red
"Medium": "#f59e0b", # amber
"Low": "#10b981", # emerald
}
# ---------------------------------------------------------------------------
# OWASP framework description maps (current 2025 IDs)
# ---------------------------------------------------------------------------
OWASP_LLM_TOP_10_DESCRIPTION_MAP = {
"LLM01": "Prompt Injection",
"LLM02": "Sensitive Information Disclosure",
"LLM03": "Supply Chain",
"LLM04": "Data and Model Poisoning",
"LLM05": "Improper Output Handling",
"LLM06": "Excessive Agency",
"LLM07": "System Prompt Leakage",
"LLM08": "Vector and Embedding Weaknesses",
"LLM09": "Misinformation",
"LLM10": "Unbounded Consumption",
}
OWASP_ML_TOP_10_DESCRIPTION_MAP = {
"ML01": "Input Manipulation Attack",
"ML02": "Data Poisoning Attack",
"ML03": "Model Inversion Attack",
"ML04": "Membership Inference Attack",
"ML05": "Model Theft",
"ML06": "AI Supply Chain Attacks",
"ML07": "Transfer Learning Attack",
"ML08": "Model Skewing",
"ML09": "Output Integrity Attack",
"ML10": "Model Poisoning",
}
OWASP_AGENTIC_TOP_10_DESCRIPTION_MAP = {
"ASI01": "Agent Goal Hijack",
"ASI02": "Tool Misuse and Exploitation",
"ASI03": "Identity and Privilege Abuse",
"ASI04": "Agentic Supply Chain Vulnerabilities",
"ASI05": "Unexpected Code Execution",
"ASI06": "Memory and Context Poisoning",
"ASI07": "Insecure Inter-Agent Communication",
"ASI08": "Cascading Failures",
"ASI09": "Human-Agent Trust Exploitation",
"ASI10": "Rogue Agents",
}
OWASP_MCP_TOP_10_DESCRIPTION_MAP = {
"MCP01": "Token Mismanagement and Secret Exposure",
"MCP02": "Privilege Escalation via Scope Creep",
"MCP03": "Tool Poisoning",
"MCP04": "Software Supply Chain Attacks and Dependency Tampering",
"MCP05": "Command Injection and Execution",
"MCP06": "Intent Flow Subversion",
"MCP07": "Insufficient Authentication and Authorization",
"MCP08": "Lack of Audit and Telemetry",
"MCP09": "Shadow MCP Servers",
"MCP10": "Context Injection and Over-Sharing",
}
pio.templates.default = "plotly_white"
st.set_page_config(
page_title="Threat Modelling of AI / ML Project",
page_icon="[AI]",
layout="wide",
initial_sidebar_state="expanded",
)
# ---------------------------------------------------------------------------
# State and selection helpers
# ---------------------------------------------------------------------------
def _type_valid(default, value):
"""Return True if value is type-compatible with default.
Rejects integers/floats saved when a selectbox stored its index instead of
its string label — the most common form of autosave corruption."""
if isinstance(default, str):
return isinstance(value, str)
if isinstance(default, list):
return isinstance(value, list)
if isinstance(default, bool):
return isinstance(value, bool)
return True # don't restrict other types
def initialize_state():
# ── Streamlit widget-GC guard ──────────────────────────────────────────────
# Streamlit removes widget keys from session_state when the widget is not
# rendered (e.g. ai_type selectbox only exists on the Project step). On
# page 3+ the key disappears and initialize_state would reset it to the
# default "Traditional ML", collapsing the whole architecture selection.
#
# Fix: we maintain shadow non-widget copies (_arch_*) that Streamlit never
# GC-s. When a widget key is missing we restore from the shadow copy before
# anything else runs.
_ARCH_KEYS = {
"ai_type": "Traditional ML",
"model_type": "Classifier",
"model_source": "Open-source",
}
for key, fallback in _ARCH_KEYS.items():
shadow = f"_arch_{key}"
if key not in st.session_state:
# Widget was GC-d — restore from shadow (or fall back to default)
st.session_state[key] = st.session_state.get(shadow, fallback)
else:
# Widget is present — refresh the shadow so it stays current
st.session_state[shadow] = st.session_state[key]
# project_name is a text_input that Streamlit GC-s whenever Step 1 is not
# rendered. Guard it with the same shadow pattern so the user's typed name
# survives navigation to other steps.
_pn_shadow = "_shadow_project_name"
if "project_name" not in st.session_state:
st.session_state["project_name"] = st.session_state.get(_pn_shadow, "")
else:
st.session_state[_pn_shadow] = st.session_state["project_name"]
# Set defaults for keys not yet in session state.
for key, default_value in initial_values.items():
if key not in st.session_state:
st.session_state[key] = default_value
elif not _type_valid(default_value, st.session_state[key]):
# Existing session-state value is wrong type (e.g. integer index
# instead of string label) — reset to the safe default.
st.session_state[key] = default_value
if "report_requested" not in st.session_state:
st.session_state.report_requested = False
if "step_index" not in st.session_state:
st.session_state.step_index = 0
# On the very first render after a server restart, restore from autosave.
# The _autosave_loaded flag ensures we only do this once per session.
if not st.session_state.get("_autosave_loaded"):
saved = _autoload()
for k, v in saved.items():
if k in initial_values: # restore form answers only
st.session_state[k] = v
# Navigation always resets to step 0 on a fresh load
st.session_state["step_index"] = 0
st.session_state["report_requested"] = False
st.session_state["_autosave_loaded"] = True
# Initialise shadow keys from the just-loaded autosave values
for key in _ARCH_KEYS:
st.session_state[f"_arch_{key}"] = st.session_state.get(key, _ARCH_KEYS[key])
# Seed the project_name shadow so it survives GC on the first render
st.session_state["_shadow_project_name"] = st.session_state.get("project_name", "")
def option_index(options, current_value, default_index=0):
return options.index(current_value) if current_value in options else default_index
def is_llm_like_selection():
return (
st.session_state.ai_type
in [
"Large Language Model (LLM)",
"Generative AI (e.g., Image/Audio Generation)",
"RAG / AI Search",
"Multimodal AI",
"Agentic AI (e.g., Autonomous Agents)",
]
or st.session_state.model_type
in [
"LLM (Generic)",
"LLM (Custom/Fine-tuned)",
"RAG Application",
"Multimodal Model",
"Agentic Workflow / Autonomous Agent",
"Multi-Agent System",
"MCP / Tool-Integrated Assistant",
]
)
def is_rag_selection():
return (
st.session_state.rag_usage == "Yes"
or st.session_state.ai_type == "RAG / AI Search"
or st.session_state.model_type in ["RAG Application", "Embedding / Vector Search"]
)
def is_multimodal_selection():
return (
st.session_state.ai_type == "Multimodal AI"
or st.session_state.model_type == "Multimodal Model"
)
def is_agentic_selection():
return (
st.session_state.ai_type == "Agentic AI (e.g., Autonomous Agents)"
or st.session_state.model_type
in ["Agentic Workflow / Autonomous Agent", "Multi-Agent System"]
)
def is_mcp_selection():
return (
st.session_state.get("mcp_usage") == "Yes"
or st.session_state.model_type == "MCP / Tool-Integrated Assistant"
)
def is_fine_tuned_selection():
return st.session_state.model_source == "Fine-tuned"
def is_trad_ml_selection():
return st.session_state.ai_type in [
"Traditional ML",
"Computer Vision",
"NLP (Non-LLM)",
"Recommendation System",
"Anomaly Detection / Fraud Detection",
] or st.session_state.model_type in [
# Names must exactly match MODEL_TYPE_OPTIONS
"Classifier",
"Regression",
"Clustering / Anomaly Detection",
"Recommender System",
"CNN / Computer Vision",
"Reinforcement Learning System",
"Edge AI / IoT Model",
]
# ---------------------------------------------------------------------------
# Adaptive questionnaire helpers
# These predicates read from st.session_state and are used to:
# - Filter options within a question (e.g. remove Garak if not LLM)
# - Conditionally show/hide questions within a step
# - Auto-clean stale answers after architecture changes
# ---------------------------------------------------------------------------
def _state_is_llm_like():
"""True if the selected AI type or model type is LLM / generative."""
return is_llm_like_selection()
def _state_is_traditional_ml():
"""True if the selected AI type is Traditional ML (classifier, regression, etc.)."""
return (
st.session_state.ai_type == "Traditional ML"
and st.session_state.model_type in [
"Classifier", "Regression", "Clustering / Anomaly Detection",
"Recommender System", "CNN / Computer Vision", "Transformer",
"Embedding / Vector Search", "Reinforcement Learning System",
"Edge AI / IoT Model", "Other Custom",
]
)
def _state_has_external_users():
"""True if end-users include customers, anonymous, or partners."""
users = st.session_state.get("users", [])
return any(u in users for u in ["Customers", "Anonymous", "Partners"])
def _state_has_sensitive_data():
"""True if data_sensitivity includes PII, PHI, financial or credential data."""
ds = st.session_state.get("data_sensitivity", [])
return any(d in ds for d in ["PII", "PHI", "Financial/payment data", "Credentials/secrets",
"Children/minors data", "Biometric data"])
def _state_has_tools_or_actions():
"""True if the system takes automatic actions or can call external tools."""
return (
st.session_state.get("plugin_access") == "Yes"
or st.session_state.get("auto_action") == "Yes"
or is_agentic_selection()
)
def _state_is_public_facing():
"""True if the service is exposed publicly or to external users."""
return st.session_state.get("exposure") in [
"Public", "Authenticated Users Only", "Partner / third-party API", "Embedded in product"
]
# Model type options filtered by the currently selected AI type
def _model_types_for_ai_type():
ai_type = st.session_state.get("ai_type", "Traditional ML")
if ai_type == "Traditional ML":
return [
"Classifier", "Regression", "Clustering / Anomaly Detection",
"Recommender System", "CNN / Computer Vision", "Transformer",
"Embedding / Vector Search", "Reinforcement Learning System",
"Edge AI / IoT Model", "Other Custom",
]
# ML sub-types — each gets a short, focused list with the best-fit
# model_type at index 0 so the default selection makes sense.
elif ai_type == "Computer Vision":
return [
"CNN / Computer Vision", "Classifier", "Transformer",
"Edge AI / IoT Model", "Other Custom",
]
elif ai_type == "NLP (Non-LLM)":
return [
"Transformer", "Classifier", "Embedding / Vector Search",
"Other Custom",
]
elif ai_type == "Recommendation System":
return [
"Recommender System", "Classifier", "Embedding / Vector Search",
"Reinforcement Learning System", "Other Custom",
]
elif ai_type == "Anomaly Detection / Fraud Detection":
return [
"Clustering / Anomaly Detection", "Classifier", "Regression",
"Other Custom",
]
elif ai_type in ["Large Language Model (LLM)", "Generative AI (e.g., Image/Audio Generation)"]:
return [
"LLM (Generic)", "LLM (Custom/Fine-tuned)", "Multimodal Model",
"Transformer", "Other Custom",
]
elif ai_type == "RAG / AI Search":
return [
"RAG Application", "Embedding / Vector Search",
"LLM (Generic)", "LLM (Custom/Fine-tuned)", "Other Custom",
]
elif ai_type == "Multimodal AI":
return [
"Multimodal Model", "LLM (Generic)", "LLM (Custom/Fine-tuned)",
"CNN / Computer Vision", "Transformer", "Other Custom",
]
elif ai_type == "Agentic AI (e.g., Autonomous Agents)":
return [
"Agentic Workflow / Autonomous Agent", "Multi-Agent System",
"MCP / Tool-Integrated Assistant", "LLM (Generic)",
"LLM (Custom/Fine-tuned)", "Other Custom",
]
return MODEL_TYPE_OPTIONS
def _adversarial_type_options():
"""Return the relevant adversarial test type options for the current AI architecture."""
opts = ["None", "Data Poisoning", "Model Evasion", "Model Inversion", "Membership Inference"]
if _state_is_llm_like():
opts += ["Prompt Injection", "Jailbreak", "Tool Misuse"]
return opts
def _adversarial_tool_options():
"""Return the relevant adversarial tool options for the current AI architecture."""
# Traditional ML-focused tools always available
opts = ["None", "CleverHans", "IBM ART", "Microsoft Counterfit", "Custom Scripts"]
if _state_is_llm_like():
# LLM-specific tools added only if LLM-like
opts = ["None", "CleverHans", "IBM ART", "Microsoft Counterfit",
"Garak", "PyRIT", "promptfoo", "Custom Scripts"]
return opts
def _clean_multiselect_answer(key, valid_options):
"""Remove any saved answers that are no longer in valid_options."""
current = st.session_state.get(key, [])
if not isinstance(current, list):
return
cleaned = [v for v in current if v in valid_options]
if cleaned != current:
st.session_state[key] = cleaned
def _outputs_options_for_arch():
"""Filter the model output options to those that make sense for the architecture.
Hides options that don't apply (e.g. Image / Audio for non-generative).
Always includes anything the user has already picked, so we never silently
drop existing answers; _clean_multiselect_answer handles the trim if needed.
"""
base = ["Text", "Labels", "Recommendations", "Decisions/scores", "Embeddings"]
if _state_is_llm_like():
base += ["Code", "API/tool calls"]
if is_generative_selection():
base += ["Image", "Audio/Video"]
if _state_has_tools_or_actions() or is_agentic_selection() or is_mcp_selection():
base += ["API/tool calls", "Files/documents", "Database writes"]
if not _state_is_llm_like() and not is_generative_selection():
# Traditional ML rarely emits Image/Audio/Code — keep list lean
base = [o for o in base if o not in {"Image", "Audio/Video"}]
# Dedupe while preserving order
seen, ordered = set(), []
for o in base:
if o not in seen:
seen.add(o); ordered.append(o)
# Always include items the user has already saved
for o in st.session_state.get("outputs", []) or []:
if o not in seen:
ordered.append(o); seen.add(o)
return ordered
def _output_destinations_options_for_arch():
"""Filter output-destination options based on what the model can actually emit.
e.g. 'Code executor' only matters if outputs include Code or API/tool calls.
Note: option labels here MUST stay in sync with the strings rules_engine.py
looks for (Database / External APIs / Files / Storage etc.). Audited
April 2026 — added 'External APIs' and 'Files / Storage' so the
'unsafe_output_persistence' rule (rules_engine.py L1875) is reachable.
"""
out = set(st.session_state.get("outputs", []) or [])
base = ["None of the above"]
# UI rendering targets
if any(o in out for o in ("Text", "Labels", "Recommendations", "Decisions/scores")) or not out:
base += ["HTML/UI", "Browser/DOM", "Email/chat message"]
# Code/tool call exec surfaces
if any(o in out for o in ("Code", "API/tool calls")) or _state_has_tools_or_actions():
base += ["Code executor", "Shell / OS", "Downstream API call", "External APIs"]
# Data writes
if "Database writes" in out or _state_has_tools_or_actions():
base += ["Database", "Files / Storage"]
if not out:
# Before the user picks outputs, show everything so they can choose
base = [
"HTML/UI", "Browser/DOM", "Code executor",
"Database", "Files / Storage", "Shell / OS",
"Downstream API call", "External APIs",
"Email/chat message", "None of the above",
]
# Dedupe
seen, ordered = set(), []
for o in base:
if o not in seen:
seen.add(o); ordered.append(o)
for o in st.session_state.get("output_destinations", []) or []:
if o not in seen:
ordered.append(o); seen.add(o)
return ordered
def is_generative_selection():
"""True if the architecture produces generative media (image/audio/text gen)."""
return st.session_state.get("ai_type") in (
"Generative AI (e.g., Image/Audio Generation)",
"Multimodal AI",
"Large Language Model (LLM)",
)
def normalize_conditional_defaults():
"""Keep conditional defaults aligned with the currently selected architecture."""
agentic = is_agentic_selection()
mcp = is_mcp_selection()
rag = is_rag_selection()
multimodal = is_multimodal_selection()
llm_like = is_llm_like_selection()
fine_tuned = is_fine_tuned_selection()
agentic_keys = [
"agentic_autonomous", "agentic_tool_access", "agentic_logging",
"agentic_hitl", "agentic_sensitive_data", "agentic_malicious_input_detection",
"agentic_memory", "agentic_memory_controls", "agentic_identity_scoped",
"agentic_code_execution", "agentic_code_sandbox", "agentic_supply_chain_controls",
"agentic_kill_switch", "agentic_multi_agent", "agentic_inter_agent_auth",
"agentic_plan_inspection",
# cross-step agentic controls
"agent_action_rate_limit", "agent_per_run_budget", "agent_execution_isolation",
"agent_scope_declared", "agent_tool_output_sanitization", "agent_destructive_action_gate",
"agent_data_write_access", "agent_exfil_controls",
]
for key in agentic_keys:
if agentic:
if st.session_state.get(key) not in YN_OPTIONS:
st.session_state[key] = "No"
else:
st.session_state[key] = "Not Applicable"
mcp_yn_keys = [
"mcp_third_party_servers", "mcp_remote_servers", "mcp_tool_schema_integrity",
"mcp_tool_output_sanitization", "mcp_authz", "mcp_human_approval",
"mcp_server_isolation",
]
# MCP step is visible when the user selected MCP *or* Agentic AI
# (agentic deployments commonly use MCP tool-servers).
mcp_step_visible = mcp or is_agentic_selection()
if mcp:
if st.session_state.get("mcp_usage") not in YN_OPTIONS:
st.session_state["mcp_usage"] = "Yes"
for key in mcp_yn_keys:
if st.session_state.get(key) not in YN_OPTIONS:
st.session_state[key] = "No"
elif mcp_step_visible:
# MCP step shown because of Agentic AI selection — user hasn't opted
# into MCP yet, so default mcp_usage to "No" (not "Not Applicable")
# so the radio widget doesn't crash. Sub-keys stay N/A until "Yes".
if st.session_state.get("mcp_usage") not in YN_OPTIONS:
st.session_state["mcp_usage"] = "No"
for key in mcp_yn_keys:
if st.session_state.get(key) not in YN_OPTIONS:
st.session_state[key] = "Not Applicable"
else:
st.session_state["mcp_usage"] = "Not Applicable"
for key in mcp_yn_keys:
st.session_state[key] = "Not Applicable"
rag_control_keys = [
"retrieval_access_control", "retrieval_content_filtering",
"vector_db_isolation", "embedding_model_provenance", "embedding_inversion_controls",
]
for key in rag_control_keys:
if rag:
if st.session_state.get(key) not in CONTROL_OPTIONS: