refactor(pep8): enforce full PEP8 compliance across backend Python codebase

- ruff.toml: select E/W/F/I/N rules, line-length=120, drop legacy ignores - Auto-fix: sort 82 import blocks (isort), remove 29 unused imports, strip 6 trailing-whitespace blank lines in docstrings - main.py: move setup_logging and settings imports to top (E402) - errors.py: noqa N818 on DDD exception names (96 call sites, safe) - intel_service.py: noqa N817 for universal ET alias - atomic/elastic/sigma import services: move _MAX_UNCOMPRESSED_SIZE and _MAX_ENTRIES to module level (N806) - compliance_import_service.py: move SAMPLE_CONTROLS / CIS_CONTROLS to module level; wrap long description strings (N806 + E501) - snapshot_service.py: move STATUS_ORDER dict to module level (N806) - sigma_import_service.py: remove dead dedup_key expression (F841) - threat_actor_import_service.py: remove dead stix_to_actor expression (F841) - data_source.py, seed_demo.py, campaign_scheduler_service.py, lolbas_import_service.py: wrap lines exceeding 120 chars (E501) - d3fend_import_service.py: per-file E501 ignore (data file with long strings) All 439 unit tests pass. ruff check app/ → All checks passed!
2026-06-09 16:40:14 +02:00
parent 1249391ef0
commit 8f98bdd273
85 changed files with 712 additions and 432 deletions
@@ -7,9 +7,9 @@ from datetime import datetime, timedelta
 from sqlalchemy import case, func
 from sqlalchemy.orm import Session

+from app.models.enums import TestResult
 from app.models.technique import Technique
 from app.models.test import Test
-from app.models.enums import TestResult


 def get_coverage_by_tactic(db: Session) -> list[dict]:
@@ -24,7 +24,6 @@ templates are identified by their ``atomic_test_id`` and simply skipped.

 import io
 import logging
-import os
 import shutil
 import tempfile
 import zipfile
@@ -54,6 +53,10 @@ _DOWNLOAD_TIMEOUT = 300
 # Top-level directory name inside the ZIP
 _ZIP_ROOT_PREFIX = "atomic-red-team-master"

+# Safety limits for ZIP extraction — prevent zip-bomb DoS
+_MAX_UNCOMPRESSED_SIZE = 500 * 1024 * 1024  # 500 MB
+_MAX_ENTRIES = 50_000
+

 # ---------------------------------------------------------------------------
 # Internal helpers
@@ -77,11 +80,6 @@ def _safe_extract_zip(zip_bytes: bytes, dest: str) -> None:
    directory (path traversal / Zip Slip) or if the archive exceeds the
    safety limits.
    """
-    # Maximum uncompressed size: 500 MB — prevents zip-bomb DoS
-    _MAX_UNCOMPRESSED_SIZE = 500 * 1024 * 1024
-    # Maximum number of entries
-    _MAX_ENTRIES = 50_000
-
    dest_path = Path(dest).resolve()

    with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
@@ -33,8 +33,8 @@ import requests as _requests
 import yaml
 from sqlalchemy.orm import Session

-from app.models.test_template import TestTemplate
 from app.models.data_source import DataSource
+from app.models.test_template import TestTemplate
 from app.services.audit_service import log_action

 logger = logging.getLogger(__name__)
@@ -16,16 +16,15 @@ from app.domain.errors import (
    PermissionViolation,
 )
 from app.models.campaign import Campaign, CampaignTest
-from app.models.test import Test
 from app.models.technique import Technique
-from app.utils import escape_like
+from app.models.test import Test
+from app.services.campaign_scheduler_service import calculate_next_run
 from app.services.campaign_service import (
+    TACTIC_TO_PHASE,
    get_campaign_progress,
    validate_no_circular_dependency,
-    TACTIC_TO_PHASE,
 )
-from app.services.campaign_scheduler_service import calculate_next_run
-
+from app.utils import escape_like

 # ── Serialization helpers ────────────────────────────────────────────────

@@ -5,17 +5,16 @@ fresh tests, and computing the next run date.
 """

 import logging
-import uuid
 from datetime import datetime, timedelta

 from sqlalchemy.orm import Session

 from app.models.campaign import Campaign, CampaignTest
-from app.models.test import Test
 from app.models.enums import TestState
-from app.services.notification_service import create_notification
-from app.services.audit_service import log_action
+from app.models.test import Test
 from app.models.user import User
+from app.services.audit_service import log_action
+from app.services.notification_service import create_notification

 logger = logging.getLogger(__name__)

@@ -166,7 +165,10 @@ def check_and_run_recurring_campaigns(db: Session) -> int:
                    user_id=campaign.created_by,
                    type="recurring_campaign_run",
                    title="Recurring campaign executed",
-                    message=f'Campaign "{child.name}" was automatically created from recurring template "{campaign.name}".',
+                    message=(
+                        f'Campaign "{child.name}" was automatically created '
+                        f'from recurring template "{campaign.name}".'
+                    ),
                    entity_type="campaign",
                    entity_id=child.id,
                )
@@ -6,18 +6,16 @@ threat actors, and progress calculation.

 import logging
 import uuid
-from datetime import datetime

 from sqlalchemy.orm import Session

 from app.domain.exceptions import EntityNotFoundError, InvalidOperationError
-from app.models.campaign import Campaign, CampaignTest, KILL_CHAIN_PHASES
+from app.models.campaign import Campaign, CampaignTest
+from app.models.enums import TechniqueStatus, TestState
+from app.models.technique import Technique
 from app.models.test import Test
 from app.models.test_template import TestTemplate
-from app.models.technique import Technique
 from app.models.threat_actor import ThreatActor, ThreatActorTechnique
-from app.models.enums import TechniqueStatus, TestState
-from app.services.notification_service import create_notification
 from app.models.user import User

 logger = logging.getLogger(__name__)
@@ -6,22 +6,256 @@ ComplianceControl, and ComplianceControlMapping records.
 """

 import logging
-import json
 import re
-from typing import Optional

 import requests
 from sqlalchemy.orm import Session

 from app.models.compliance import (
-    ComplianceFramework,
    ComplianceControl,
    ComplianceControlMapping,
+    ComplianceFramework,
 )
 from app.models.technique import Technique

 logger = logging.getLogger(__name__)

+# ── Module-level control definitions (avoids N806 / uppercase-in-function) ────
+
+_NIST_SAMPLE_CONTROLS = [
+    {
+        "control_id": "AC-2",
+        "title": "Account Management",
+        "category": "Access Control",
+        "techniques": ["T1078", "T1136", "T1098", "T1087", "T1069"],
+    },
+    {
+        "control_id": "AC-3",
+        "title": "Access Enforcement",
+        "category": "Access Control",
+        "techniques": ["T1078", "T1548", "T1134"],
+    },
+    {
+        "control_id": "AC-4",
+        "title": "Information Flow Enforcement",
+        "category": "Access Control",
+        "techniques": ["T1048", "T1041", "T1572"],
+    },
+    {
+        "control_id": "AC-6",
+        "title": "Least Privilege",
+        "category": "Access Control",
+        "techniques": ["T1078", "T1548", "T1134"],
+    },
+    {
+        "control_id": "AU-2",
+        "title": "Event Logging",
+        "category": "Audit and Accountability",
+        "techniques": ["T1562", "T1070"],
+    },
+    {
+        "control_id": "AU-6",
+        "title": "Audit Record Review",
+        "category": "Audit and Accountability",
+        "techniques": ["T1562", "T1070", "T1027"],
+    },
+    {
+        "control_id": "CA-7",
+        "title": "Continuous Monitoring",
+        "category": "Assessment, Authorization, and Monitoring",
+        "techniques": ["T1059", "T1053"],
+    },
+    {
+        "control_id": "CM-2",
+        "title": "Baseline Configuration",
+        "category": "Configuration Management",
+        "techniques": ["T1574", "T1546"],
+    },
+    {
+        "control_id": "CM-6",
+        "title": "Configuration Settings",
+        "category": "Configuration Management",
+        "techniques": ["T1574", "T1546", "T1112"],
+    },
+    {
+        "control_id": "CM-7",
+        "title": "Least Functionality",
+        "category": "Configuration Management",
+        "techniques": ["T1059", "T1218"],
+    },
+    {
+        "control_id": "IA-2",
+        "title": "Identification and Authentication",
+        "category": "Identification and Authentication",
+        "techniques": ["T1078", "T1110"],
+    },
+    {
+        "control_id": "IA-5",
+        "title": "Authenticator Management",
+        "category": "Identification and Authentication",
+        "techniques": ["T1078", "T1110", "T1003"],
+    },
+    {
+        "control_id": "IR-4",
+        "title": "Incident Handling",
+        "category": "Incident Response",
+        "techniques": ["T1059", "T1547"],
+    },
+    {
+        "control_id": "RA-5",
+        "title": "Vulnerability Monitoring and Scanning",
+        "category": "Risk Assessment",
+        "techniques": ["T1190", "T1203"],
+    },
+    {
+        "control_id": "SC-7",
+        "title": "Boundary Protection",
+        "category": "System and Communications Protection",
+        "techniques": ["T1048", "T1041", "T1071"],
+    },
+    {
+        "control_id": "SC-28",
+        "title": "Protection of Information at Rest",
+        "category": "System and Communications Protection",
+        "techniques": ["T1005", "T1114"],
+    },
+    {
+        "control_id": "SI-3",
+        "title": "Malicious Code Protection",
+        "category": "System and Information Integrity",
+        "techniques": ["T1059", "T1204", "T1566"],
+    },
+    {
+        "control_id": "SI-4",
+        "title": "System Monitoring",
+        "category": "System and Information Integrity",
+        "techniques": ["T1059", "T1053", "T1547"],
+    },
+    {
+        "control_id": "SI-7",
+        "title": "Software, Firmware, and Information Integrity",
+        "category": "System and Information Integrity",
+        "techniques": ["T1195", "T1553"],
+    },
+    {
+        "control_id": "PM-16",
+        "title": "Threat Awareness Program",
+        "category": "Program Management",
+        "techniques": ["T1566", "T1204"],
+    },
+]
+
+_CIS_CONTROLS = [
+    {
+        "control_id": "CIS-1",
+        "title": "Inventory and Control of Enterprise Assets",
+        "category": "IG1 — Basic",
+        "techniques": ["T1595", "T1590", "T1018", "T1082"],
+    },
+    {
+        "control_id": "CIS-2",
+        "title": "Inventory and Control of Software Assets",
+        "category": "IG1 — Basic",
+        "techniques": ["T1518", "T1072", "T1195"],
+    },
+    {
+        "control_id": "CIS-3",
+        "title": "Data Protection",
+        "category": "IG1 — Basic",
+        "techniques": ["T1005", "T1114", "T1560", "T1048", "T1041"],
+    },
+    {
+        "control_id": "CIS-4",
+        "title": "Secure Configuration of Enterprise Assets and Software",
+        "category": "IG1 — Basic",
+        "techniques": ["T1574", "T1546", "T1112", "T1543"],
+    },
+    {
+        "control_id": "CIS-5",
+        "title": "Account Management",
+        "category": "IG1 — Basic",
+        "techniques": ["T1078", "T1136", "T1098", "T1087"],
+    },
+    {
+        "control_id": "CIS-6",
+        "title": "Access Control Management",
+        "category": "IG1 — Basic",
+        "techniques": ["T1078", "T1548", "T1134", "T1021"],
+    },
+    {
+        "control_id": "CIS-7",
+        "title": "Continuous Vulnerability Management",
+        "category": "IG2 — Foundational",
+        "techniques": ["T1190", "T1203", "T1068", "T1210"],
+    },
+    {
+        "control_id": "CIS-8",
+        "title": "Audit Log Management",
+        "category": "IG2 — Foundational",
+        "techniques": ["T1562", "T1070", "T1059"],
+    },
+    {
+        "control_id": "CIS-9",
+        "title": "Email and Web Browser Protections",
+        "category": "IG2 — Foundational",
+        "techniques": ["T1566", "T1204", "T1189", "T1598"],
+    },
+    {
+        "control_id": "CIS-10",
+        "title": "Malware Defenses",
+        "category": "IG2 — Foundational",
+        "techniques": ["T1059", "T1204", "T1027", "T1140", "T1497"],
+    },
+    {
+        "control_id": "CIS-11",
+        "title": "Data Recovery",
+        "category": "IG1 — Basic",
+        "techniques": ["T1486", "T1490", "T1561"],
+    },
+    {
+        "control_id": "CIS-12",
+        "title": "Network Infrastructure Management",
+        "category": "IG2 — Foundational",
+        "techniques": ["T1557", "T1071", "T1572", "T1571"],
+    },
+    {
+        "control_id": "CIS-13",
+        "title": "Network Monitoring and Defense",
+        "category": "IG2 — Foundational",
+        "techniques": ["T1071", "T1048", "T1041", "T1105", "T1572"],
+    },
+    {
+        "control_id": "CIS-14",
+        "title": "Security Awareness and Skills Training",
+        "category": "IG1 — Basic",
+        "techniques": ["T1566", "T1204", "T1598"],
+    },
+    {
+        "control_id": "CIS-15",
+        "title": "Service Provider Management",
+        "category": "IG2 — Foundational",
+        "techniques": ["T1199", "T1195"],
+    },
+    {
+        "control_id": "CIS-16",
+        "title": "Application Software Security",
+        "category": "IG2 — Foundational",
+        "techniques": ["T1190", "T1059", "T1203"],
+    },
+    {
+        "control_id": "CIS-17",
+        "title": "Incident Response Management",
+        "category": "IG2 — Foundational",
+        "techniques": ["T1059", "T1547", "T1053"],
+    },
+    {
+        "control_id": "CIS-18",
+        "title": "Penetration Testing",
+        "category": "IG3 — Organizational",
+        "techniques": ["T1595", "T1046", "T1190", "T1059"],
+    },
+]
+
 # URL for the NIST 800-53 Rev 5 to ATT&CK mapping
 # This is the JSON STIX bundle that contains the relationships
 NIST_MAPPING_URL = (
@@ -53,7 +287,11 @@ def import_nist_800_53_mappings(db: Session) -> dict:
        framework = ComplianceFramework(
            name="NIST 800-53 Rev 5",
            version="5",
-            description="National Institute of Standards and Technology Special Publication 800-53 Revision 5 — Security and Privacy Controls for Information Systems and Organizations",
+            description=(
+                "National Institute of Standards and Technology "
+                "Special Publication 800-53 Revision 5 — "
+                "Security and Privacy Controls for Information Systems and Organizations"
+            ),
            url="https://csrc.nist.gov/publications/detail/sp/800-53/rev-5/final",
            is_active=True,
        )
@@ -216,49 +454,6 @@ def _import_sample_nist_mappings(db: Session, framework: ComplianceFramework) ->

    This ensures the feature works even without network access.
    """
-    SAMPLE_CONTROLS = [
-        {"control_id": "AC-2", "title": "Account Management", "category": "Access Control",
-         "techniques": ["T1078", "T1136", "T1098", "T1087", "T1069"]},
-        {"control_id": "AC-3", "title": "Access Enforcement", "category": "Access Control",
-         "techniques": ["T1078", "T1548", "T1134"]},
-        {"control_id": "AC-4", "title": "Information Flow Enforcement", "category": "Access Control",
-         "techniques": ["T1048", "T1041", "T1572"]},
-        {"control_id": "AC-6", "title": "Least Privilege", "category": "Access Control",
-         "techniques": ["T1078", "T1548", "T1134"]},
-        {"control_id": "AU-2", "title": "Event Logging", "category": "Audit and Accountability",
-         "techniques": ["T1562", "T1070"]},
-        {"control_id": "AU-6", "title": "Audit Record Review", "category": "Audit and Accountability",
-         "techniques": ["T1562", "T1070", "T1027"]},
-        {"control_id": "CA-7", "title": "Continuous Monitoring", "category": "Assessment, Authorization, and Monitoring",
-         "techniques": ["T1059", "T1053"]},
-        {"control_id": "CM-2", "title": "Baseline Configuration", "category": "Configuration Management",
-         "techniques": ["T1574", "T1546"]},
-        {"control_id": "CM-6", "title": "Configuration Settings", "category": "Configuration Management",
-         "techniques": ["T1574", "T1546", "T1112"]},
-        {"control_id": "CM-7", "title": "Least Functionality", "category": "Configuration Management",
-         "techniques": ["T1059", "T1218"]},
-        {"control_id": "IA-2", "title": "Identification and Authentication", "category": "Identification and Authentication",
-         "techniques": ["T1078", "T1110"]},
-        {"control_id": "IA-5", "title": "Authenticator Management", "category": "Identification and Authentication",
-         "techniques": ["T1078", "T1110", "T1003"]},
-        {"control_id": "IR-4", "title": "Incident Handling", "category": "Incident Response",
-         "techniques": ["T1059", "T1547"]},
-        {"control_id": "RA-5", "title": "Vulnerability Monitoring and Scanning", "category": "Risk Assessment",
-         "techniques": ["T1190", "T1203"]},
-        {"control_id": "SC-7", "title": "Boundary Protection", "category": "System and Communications Protection",
-         "techniques": ["T1048", "T1041", "T1071"]},
-        {"control_id": "SC-28", "title": "Protection of Information at Rest", "category": "System and Communications Protection",
-         "techniques": ["T1005", "T1114"]},
-        {"control_id": "SI-3", "title": "Malicious Code Protection", "category": "System and Information Integrity",
-         "techniques": ["T1059", "T1204", "T1566"]},
-        {"control_id": "SI-4", "title": "System Monitoring", "category": "System and Information Integrity",
-         "techniques": ["T1059", "T1053", "T1547"]},
-        {"control_id": "SI-7", "title": "Software, Firmware, and Information Integrity", "category": "System and Information Integrity",
-         "techniques": ["T1195", "T1553"]},
-        {"control_id": "PM-16", "title": "Threat Awareness Program", "category": "Program Management",
-         "techniques": ["T1566", "T1204"]},
-    ]
-
    # Build technique lookup
    all_techniques = {t.mitre_id: t for t in db.query(Technique).all()}

@@ -276,7 +471,7 @@ def _import_sample_nist_mappings(db: Session, framework: ComplianceFramework) ->
    controls_created = 0
    mappings_created = 0

-    for sample in SAMPLE_CONTROLS:
+    for sample in _NIST_SAMPLE_CONTROLS:
        # Create or get control
        if sample["control_id"] in existing_controls:
            control = existing_controls[sample["control_id"]]
@@ -348,8 +543,11 @@ def import_cis_controls_v8_mappings(db: Session) -> dict:
        framework = ComplianceFramework(
            name="CIS Controls v8",
            version="8",
-            description="Center for Internet Security Critical Security Controls Version 8 — "
-                        "a prioritized set of 18 security safeguards organized by Implementation Groups (IG1, IG2, IG3).",
+            description=(
+                "Center for Internet Security Critical Security Controls Version 8 — "
+                "a prioritized set of 18 security safeguards "
+                "organized by Implementation Groups (IG1, IG2, IG3)."
+            ),
            url="https://www.cisecurity.org/controls/v8",
            is_active=True,
        )
@@ -360,62 +558,7 @@ def import_cis_controls_v8_mappings(db: Session) -> dict:
        logger.info("CIS Controls v8 framework already exists")

    # ── 2. Control definitions with ATT&CK mappings ───────────────
-    CIS_CONTROLS = [
-        {"control_id": "CIS-1", "title": "Inventory and Control of Enterprise Assets",
-         "category": "IG1 — Basic",
-         "techniques": ["T1595", "T1590", "T1018", "T1082"]},
-        {"control_id": "CIS-2", "title": "Inventory and Control of Software Assets",
-         "category": "IG1 — Basic",
-         "techniques": ["T1518", "T1072", "T1195"]},
-        {"control_id": "CIS-3", "title": "Data Protection",
-         "category": "IG1 — Basic",
-         "techniques": ["T1005", "T1114", "T1560", "T1048", "T1041"]},
-        {"control_id": "CIS-4", "title": "Secure Configuration of Enterprise Assets and Software",
-         "category": "IG1 — Basic",
-         "techniques": ["T1574", "T1546", "T1112", "T1543"]},
-        {"control_id": "CIS-5", "title": "Account Management",
-         "category": "IG1 — Basic",
-         "techniques": ["T1078", "T1136", "T1098", "T1087"]},
-        {"control_id": "CIS-6", "title": "Access Control Management",
-         "category": "IG1 — Basic",
-         "techniques": ["T1078", "T1548", "T1134", "T1021"]},
-        {"control_id": "CIS-7", "title": "Continuous Vulnerability Management",
-         "category": "IG2 — Foundational",
-         "techniques": ["T1190", "T1203", "T1068", "T1210"]},
-        {"control_id": "CIS-8", "title": "Audit Log Management",
-         "category": "IG2 — Foundational",
-         "techniques": ["T1562", "T1070", "T1059"]},
-        {"control_id": "CIS-9", "title": "Email and Web Browser Protections",
-         "category": "IG2 — Foundational",
-         "techniques": ["T1566", "T1204", "T1189", "T1598"]},
-        {"control_id": "CIS-10", "title": "Malware Defenses",
-         "category": "IG2 — Foundational",
-         "techniques": ["T1059", "T1204", "T1027", "T1140", "T1497"]},
-        {"control_id": "CIS-11", "title": "Data Recovery",
-         "category": "IG1 — Basic",
-         "techniques": ["T1486", "T1490", "T1561"]},
-        {"control_id": "CIS-12", "title": "Network Infrastructure Management",
-         "category": "IG2 — Foundational",
-         "techniques": ["T1557", "T1071", "T1572", "T1571"]},
-        {"control_id": "CIS-13", "title": "Network Monitoring and Defense",
-         "category": "IG2 — Foundational",
-         "techniques": ["T1071", "T1048", "T1041", "T1105", "T1572"]},
-        {"control_id": "CIS-14", "title": "Security Awareness and Skills Training",
-         "category": "IG1 — Basic",
-         "techniques": ["T1566", "T1204", "T1598"]},
-        {"control_id": "CIS-15", "title": "Service Provider Management",
-         "category": "IG2 — Foundational",
-         "techniques": ["T1199", "T1195"]},
-        {"control_id": "CIS-16", "title": "Application Software Security",
-         "category": "IG2 — Foundational",
-         "techniques": ["T1190", "T1059", "T1203"]},
-        {"control_id": "CIS-17", "title": "Incident Response Management",
-         "category": "IG2 — Foundational",
-         "techniques": ["T1059", "T1547", "T1053"]},
-        {"control_id": "CIS-18", "title": "Penetration Testing",
-         "category": "IG3 — Organizational",
-         "techniques": ["T1595", "T1046", "T1190", "T1059"]},
-    ]
+    # (defined at module level as _CIS_CONTROLS)

    # Build technique lookup
    all_techniques = {t.mitre_id: t for t in db.query(Technique).all()}
@@ -439,7 +582,7 @@ def import_cis_controls_v8_mappings(db: Session) -> dict:
    controls_created = 0
    mappings_created = 0

-    for item in CIS_CONTROLS:
+    for item in _CIS_CONTROLS:
        if item["control_id"] in existing_controls:
            control = existing_controls[item["control_id"]]
        else:
@@ -16,16 +16,15 @@ from sqlalchemy.orm import Session

 from app.domain.errors import EntityNotFoundError
 from app.models.compliance import (
-    ComplianceFramework,
    ComplianceControl,
    ComplianceControlMapping,
+    ComplianceFramework,
 )
 from app.models.technique import Technique
 from app.models.test_template import TestTemplate
 from app.models.threat_actor import ThreatActorTechnique
 from app.services.scoring_service import calculate_technique_score

-
 # ── Helpers ───────────────────────────────────────────────────────────


@@ -7,14 +7,13 @@ Uses the D3FEND public API:
 """

 import logging
-import uuid
 from typing import Any

 import httpx
 from sqlalchemy.orm import Session

-from app.models.technique import Technique
 from app.models.defensive_technique import DefensiveTechnique, DefensiveTechniqueMapping
+from app.models.technique import Technique

 logger = logging.getLogger(__name__)

@@ -405,9 +404,10 @@ def sync(db: Session) -> dict:
    Called by the Data Sources router when the user clicks Sync for D3FEND.
    Returns a flat summary dict suitable for ``last_sync_stats``.
    """
-    from app.models.data_source import DataSource
    from datetime import datetime

+    from app.models.data_source import DataSource
+
    tech_result = import_d3fend_techniques(db)
    mapping_result = import_d3fend_mappings(db)

@@ -164,8 +164,8 @@ def get_source_stats(db: Session, source_id: str) -> dict:
    if not ds:
        raise EntityNotFoundError("Data source", source_id)

-    from app.models.test_template import TestTemplate
    from app.models.detection_rule import DetectionRule
+    from app.models.test_template import TestTemplate

    template_count = 0
    rule_count = 0
@@ -15,14 +15,13 @@ from sqlalchemy.orm import Session

 from app.domain.errors import EntityNotFoundError
 from app.models.detection_rule import DetectionRule
+from app.models.technique import Technique
 from app.models.test import Test
+from app.models.test_detection_result import TestDetectionResult
 from app.models.test_template import TestTemplate
 from app.models.test_template_detection_rule import TestTemplateDetectionRule
-from app.models.test_detection_result import TestDetectionResult
-from app.models.technique import Technique
 from app.utils import escape_like

-
 # ── Public service functions ──────────────────────────────────────────


@@ -32,8 +32,8 @@ from pathlib import Path
 import requests as _requests
 from sqlalchemy.orm import Session

-from app.models.detection_rule import DetectionRule
 from app.models.data_source import DataSource
+from app.models.detection_rule import DetectionRule
 from app.services.audit_service import log_action

 logger = logging.getLogger(__name__)
@@ -50,6 +50,10 @@ ELASTIC_ZIP_URL = (
 _DOWNLOAD_TIMEOUT = 300
 _ZIP_ROOT_PREFIX = "detection-rules-main"

+# Safety limits for ZIP extraction — prevent zip-bomb DoS
+_MAX_UNCOMPRESSED_SIZE = 500 * 1024 * 1024  # 500 MB
+_MAX_ENTRIES = 50_000
+
 # Severity normalisation
 _SEVERITY_MAP = {
    "informational": "informational",
@@ -82,11 +86,6 @@ def _safe_extract_zip(zip_bytes: bytes, dest: str) -> None:
    directory (path traversal / Zip Slip) or if the archive exceeds the
    safety limits.
    """
-    # Maximum uncompressed size: 500 MB — prevents zip-bomb DoS
-    _MAX_UNCOMPRESSED_SIZE = 500 * 1024 * 1024
-    # Maximum number of entries
-    _MAX_ENTRIES = 50_000
-
    dest_path = Path(dest).resolve()

    with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
@@ -10,7 +10,6 @@ no ``db.commit()``.
 from __future__ import annotations

 import json
-from typing import Optional

 from sqlalchemy import func, or_
 from sqlalchemy.orm import Session
@@ -18,7 +17,6 @@ from sqlalchemy.orm import Session
 from app.domain.errors import BusinessRuleViolation, EntityNotFoundError
 from app.models.campaign import Campaign, CampaignTest
 from app.models.detection_rule import DetectionRule
-from app.models.defensive_technique import DefensiveTechniqueMapping
 from app.models.enums import TechniqueStatus, TestState
 from app.models.technique import Technique
 from app.models.test import Test
@@ -11,9 +11,9 @@ parser.  No LLMs or paid APIs are used.

 import logging
 import re
-import defusedxml.ElementTree as ET
 from datetime import datetime

+import defusedxml.ElementTree as ET  # noqa: N817 — ET is the universal stdlib alias for ElementTree
 import requests as _requests
 from sqlalchemy.orm import Session

@@ -37,8 +37,8 @@ import requests as _requests
 import yaml
 from sqlalchemy.orm import Session

-from app.models.test_template import TestTemplate
 from app.models.data_source import DataSource
+from app.models.test_template import TestTemplate
 from app.services.audit_service import log_action

 logger = logging.getLogger(__name__)
@@ -176,7 +176,11 @@ def _parse_lolbas(root_dir: Path) -> list[dict]:
            results.append({
                "mitre_technique_id": mitre_id,
                "name": f"LOLBAS: {binary_name} — {usecase or cmd_description or mitre_id}"[:500],
-                "description": f"{description}\n\n{cmd_description}".strip()[:2000] if description else cmd_description[:2000] if cmd_description else None,
+                "description": (
+                    f"{description}\n\n{cmd_description}".strip()[:2000]
+                    if description
+                    else cmd_description[:2000] if cmd_description else None
+                ),
                "source": "lolbas",
                "platform": "windows",
                "tool_suggested": binary_name,
@@ -13,8 +13,8 @@ import requests as _requests
 from sqlalchemy.orm import Session
 from taxii2client.v20 import Server as TaxiiServer

-from app.models.technique import Technique
 from app.models.enums import TechniqueStatus
+from app.models.technique import Technique
 from app.services.audit_service import log_action

 logger = logging.getLogger(__name__)
@@ -10,14 +10,13 @@ but do **not** commit.  The caller is responsible for committing.
 import uuid
 from datetime import datetime, timedelta

-from sqlalchemy.orm import Session
 from sqlalchemy import func
+from sqlalchemy.orm import Session

 from app.domain.errors import EntityNotFoundError
 from app.models.notification import Notification
 from app.models.user import User

-
 # ---------------------------------------------------------------------------
 # Core CRUD
 # ---------------------------------------------------------------------------
@@ -6,14 +6,14 @@ Calculates security operations KPIs from test data and audit logs.
 from datetime import datetime, timedelta
 from typing import Optional

-from sqlalchemy import func, case, and_, or_, extract
+from sqlalchemy import func
 from sqlalchemy.orm import Session

-from app.models.test import Test
-from app.models.technique import Technique
-from app.models.test_detection_result import TestDetectionResult
 from app.models.audit import AuditLog
-from app.models.enums import TestState, TestResult
+from app.models.enums import TestResult, TestState
+from app.models.technique import Technique
+from app.models.test import Test
+from app.models.test_detection_result import TestDetectionResult


 def _safe_stats(values: list[float]) -> dict:
@@ -3,9 +3,9 @@
 Uses WeasyPrint for PDF generation and docxtpl for DOCX.
 """

+import logging
 import os
 import uuid
-import logging
 from datetime import datetime

 from jinja2 import Environment, FileSystemLoader
@@ -34,7 +34,7 @@ class ReportEngine:

    def generate_pdf(self, template_name: str, context: dict) -> str:
        """Render HTML and convert to PDF with WeasyPrint."""
-        from weasyprint import HTML, CSS
+        from weasyprint import CSS, HTML

        html_content = self.render_html(template_name, context)
        css_path = os.path.join(settings.REPORT_TEMPLATES_DIR, "styles", "report.css")
@@ -98,7 +98,7 @@ def generate_coverage_report(
    output_format: str = "pdf",
 ) -> str:
    """Generate an organization-wide MITRE ATT&CK coverage report."""
-    from sqlalchemy import func, case
+    from sqlalchemy import case, func

    org_score = _safe_org_score(db)

@@ -234,7 +234,8 @@ def generate_quarterly_summary(
    output_format: str = "pdf",
 ) -> str:
    """Quarterly summary — reuses executive metrics plus snapshot trend rows."""
-    from sqlalchemy import case as sql_case, func
+    from sqlalchemy import case as sql_case
+    from sqlalchemy import func

    org_score = _safe_org_score(db)
    quarter_ago = datetime.utcnow() - timedelta(days=90)
@@ -58,13 +58,13 @@ def get_organization_score_cached(db):
 def get_operational_metrics_cached(db):
    """Cached wrapper around operational metrics (MTTD, MTTR, efficacy)."""
    from app.services.operational_metrics_service import (
-        calculate_mttd,
-        calculate_mttr,
-        calculate_detection_efficacy,
        calculate_alert_fidelity,
        calculate_coverage_velocity,
-        calculate_validation_throughput,
+        calculate_detection_efficacy,
+        calculate_mttd,
+        calculate_mttr,
        calculate_rejection_rate,
+        calculate_validation_throughput,
    )

    cached = get("op_metrics")
@@ -10,19 +10,18 @@ never produce N+1 traffic.
 """

 from datetime import datetime, timedelta, timezone
-from typing import Optional

 from sqlalchemy import case, func
 from sqlalchemy.orm import Session

 from app.domain.errors import EntityNotFoundError
+from app.models.defensive_technique import DefensiveTechniqueMapping
+from app.models.detection_rule import DetectionRule
+from app.models.enums import TestResult, TestState
 from app.models.technique import Technique
 from app.models.test import Test
-from app.models.detection_rule import DetectionRule
 from app.models.test_detection_result import TestDetectionResult
-from app.models.defensive_technique import DefensiveTechniqueMapping
 from app.models.threat_actor import ThreatActor, ThreatActorTechnique
-from app.models.enums import TestState, TestResult
 from app.services.scoring_config_service import get_scoring_weights

 _SEVERITY_FACTORS: dict[str, float] = {
@@ -659,7 +658,6 @@ def get_score_history(db: Session, period: str = "90d") -> list:
    computing scores based on test dates within time windows.
    Returns a list of weekly data points.
    """
-    from app.models.audit import AuditLog

    now = datetime.utcnow()
    if period == "30d":
@@ -35,8 +35,8 @@ import requests as _requests
 import yaml
 from sqlalchemy.orm import Session

-from app.models.detection_rule import DetectionRule
 from app.models.data_source import DataSource
+from app.models.detection_rule import DetectionRule
 from app.services.audit_service import log_action

 logger = logging.getLogger(__name__)
@@ -52,6 +52,10 @@ SIGMA_ZIP_URL = (
 _DOWNLOAD_TIMEOUT = 300
 _ZIP_ROOT_PREFIX = "sigma-master"

+# Safety limits for ZIP extraction — prevent zip-bomb DoS
+_MAX_UNCOMPRESSED_SIZE = 500 * 1024 * 1024  # 500 MB
+_MAX_ENTRIES = 50_000
+
 # Regex to extract MITRE ATT&CK technique IDs from Sigma tags
 # e.g. "attack.t1059.001" → "T1059.001"
 _ATTACK_TAG_RE = re.compile(r"attack\.(t\d{4}(?:\.\d{3})?)", re.IGNORECASE)
@@ -88,11 +92,6 @@ def _safe_extract_zip(zip_bytes: bytes, dest: str) -> None:
    directory (path traversal / Zip Slip) or if the archive exceeds the
    safety limits.
    """
-    # Maximum uncompressed size: 500 MB — prevents zip-bomb DoS
-    _MAX_UNCOMPRESSED_SIZE = 500 * 1024 * 1024
-    # Maximum number of entries
-    _MAX_ENTRIES = 50_000
-
    dest_path = Path(dest).resolve()

    with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
@@ -290,11 +289,8 @@ def sync(db: Session) -> dict:
    skipped = 0

    for item in parsed_rules:
-        # Dedup key: source_id (relative path). A rule file may produce
-        # multiple entries (one per technique), but we deduplicate by
-        # source_id so re-runs are safe.  For multi-technique rules we
-        # only skip if the exact same source_id is already present.
-        dedup_key = f"{item['source_id']}::{item['mitre_technique_id']}"
+        # Deduplicate by source_id: one rule file may map to multiple techniques,
+        # but we skip insertion if this source_id was already imported.
        if item["source_id"] in existing_ids:
            skipped += 1
            continue
@@ -16,9 +16,9 @@ from sqlalchemy import func
 from sqlalchemy.orm import Session

 from app.domain.errors import EntityNotFoundError
-from app.models.technique import Technique
 from app.models.coverage_snapshot import CoverageSnapshot, SnapshotTechniqueState
 from app.models.enums import TechniqueStatus
+from app.models.technique import Technique
 from app.services.scoring_service import (
    bulk_technique_scores,
    calculate_organization_score,
@@ -26,6 +26,15 @@ from app.services.scoring_service import (

 logger = logging.getLogger(__name__)

+# Coverage status ordering for snapshot delta comparisons (higher = better coverage)
+_STATUS_ORDER: dict[str, int] = {
+    "not_evaluated": 0,
+    "not_covered": 1,
+    "in_progress": 2,
+    "partial": 3,
+    "validated": 4,
+}
+

 # ---------------------------------------------------------------------------
 # Serialization and queries
@@ -296,15 +305,6 @@ def compare_snapshots(
        .all()
    }

-    # Status priority for comparison
-    STATUS_ORDER = {
-        "not_evaluated": 0,
-        "not_covered": 1,
-        "in_progress": 2,
-        "partial": 3,
-        "validated": 4,
-    }
-
    improved = []
    worsened = []
    unchanged_count = 0
@@ -315,8 +315,8 @@ def compare_snapshots(
        a = states_a.get(mitre_id, {"status": "not_evaluated", "score": 0})
        b = states_b.get(mitre_id, {"status": "not_evaluated", "score": 0})

-        a_order = STATUS_ORDER.get(a["status"], 0)
-        b_order = STATUS_ORDER.get(b["status"], 0)
+        a_order = _STATUS_ORDER.get(a["status"], 0)
+        b_order = _STATUS_ORDER.get(b["status"], 0)

        if b_order > a_order or (b_order == a_order and b["score"] > a["score"]):
            improved.append({
@@ -14,11 +14,11 @@ from app.domain.errors import (
    EntityNotFoundError,
    PermissionViolation,
 )
+from app.models.audit import AuditLog
 from app.models.enums import TestState
 from app.models.technique import Technique
 from app.models.test import Test
 from app.models.test_template import TestTemplate
-from app.models.audit import AuditLog
 from app.utils import escape_like


@@ -18,13 +18,16 @@ from datetime import datetime
 from sqlalchemy.orm import Session

 from app.config import settings
-from app.domain.exceptions import InvalidOperationError, InvalidTransitionError
+from app.domain.exceptions import InvalidOperationError
 from app.domain.test_entity import TestEntity
 from app.models.enums import TestState
 from app.models.test import Test
 from app.models.user import User
 from app.services.audit_service import log_action
-from app.services.notification_service import notify_test_state_change, create_notification
+from app.services.notification_service import (
+    create_notification,
+    notify_test_state_change,
+)

 logger = logging.getLogger(__name__)

@@ -38,9 +38,9 @@ from pathlib import Path
 import requests as _requests
 from sqlalchemy.orm import Session

-from app.models.threat_actor import ThreatActor, ThreatActorTechnique
-from app.models.technique import Technique
 from app.models.data_source import DataSource
+from app.models.technique import Technique
+from app.models.threat_actor import ThreatActor, ThreatActorTechnique
 from app.services.audit_service import log_action

 logger = logging.getLogger(__name__)
@@ -241,9 +241,6 @@ def sync(db: Session) -> dict:
    relationships = _parse_relationships(objects)
    attack_pattern_map = _build_attack_pattern_map(objects)

-    # Step 2: Build STIX-ID → actor dict map
-    stix_to_actor = {a["stix_id"]: a for a in actor_dicts}
-
    # Step 3: Load existing actors and techniques from DB
    existing_actors = {
        row.mitre_id: row
@@ -15,13 +15,12 @@ from sqlalchemy.orm import Session

 from app.domain.errors import EntityNotFoundError
 from app.models.enums import TechniqueStatus
+from app.models.technique import Technique
 from app.models.test import Test
 from app.models.test_template import TestTemplate
 from app.models.threat_actor import ThreatActor, ThreatActorTechnique
-from app.models.technique import Technique
 from app.utils import escape_like

-
 # ── Public service functions ──────────────────────────────────────────


@@ -11,7 +11,11 @@ import uuid
 from sqlalchemy.orm import Session

 from app.auth import hash_password
-from app.domain.errors import BusinessRuleViolation, DuplicateEntityError, EntityNotFoundError
+from app.domain.errors import (
+    BusinessRuleViolation,
+    DuplicateEntityError,
+    EntityNotFoundError,
+)
 from app.models.user import User

 VALID_ROLES = {"admin", "red_tech", "blue_tech", "red_lead", "blue_lead", "viewer"}