more linting, minor enrichment skill update

Author: Lee Chagolla-Christensen

.claude/skills/enrichment-module-builder.md

@@ -423,6 +423,12 @@ class Test{ModuleName}Analyzer:
             # Assert on results
 ```
 
+Guidelines:
+- Test both positive(happy path) and negative(unhappy path) cases.
+  - Do this for all result types, findings, transforms, DB writes, and file uploads generated.
+- Handle edge cases and error conditions
+
+
 ### Run Tests
 
 ```bash

libs/chromium/chromium/helpers.py

@@ -295,7 +295,7 @@ def is_valid_text(data: bytes) -> bool:
 
 
 def byte_xor(ba1, ba2):
-    return bytes([_a ^ _b for _a, _b in zip(ba1, ba2)])
+    return bytes([_a ^ _b for _a, _b in zip(ba1, ba2, strict=False)])
 
 
 def parse_abe_blob(abe_data: bytes, chromekey: bytes | None = None) -> dict | None:

libs/common/common/models2/dpapi.py

@@ -1,7 +1,7 @@
 """DPAPI credential models for API requests."""
 
 import re
-from typing import Annotated, Literal, Union
+from typing import Annotated, Literal
 from uuid import UUID
 
 from common.logger import get_logger
@@ -201,15 +201,13 @@ def get_credential_type(v):
 
 
 type DpapiCredentialRequest = Annotated[
-    Union[
-        Annotated[PasswordCredentialKey, Tag("password")],
-        Annotated[NtlmHashCredentialKey, Tag("cred_key_ntlm")],
-        Annotated[Sha1CredentialKey, Tag("cred_key_sha1")],
-        Annotated[Pbkdf2StrongCredentialKey, Tag("cred_key_pbkdf2")],
-        Annotated[DomainBackupKeyCredential, Tag("domain_backup_key")],
-        Annotated[MasterKeyGuidPairList, Tag("master_key_guid_pair")],
-        Annotated[DpapiSystemCredentialRequest, Tag("dpapi_system")],
-        Annotated[ChromiumAppBoundKeyCredential, Tag("chromium_app_bound_key")],
-    ],
+    Annotated[PasswordCredentialKey, Tag("password")]
+    | Annotated[NtlmHashCredentialKey, Tag("cred_key_ntlm")]
+    | Annotated[Sha1CredentialKey, Tag("cred_key_sha1")]
+    | Annotated[Pbkdf2StrongCredentialKey, Tag("cred_key_pbkdf2")]
+    | Annotated[DomainBackupKeyCredential, Tag("domain_backup_key")]
+    | Annotated[MasterKeyGuidPairList, Tag("master_key_guid_pair")]
+    | Annotated[DpapiSystemCredentialRequest, Tag("dpapi_system")]
+    | Annotated[ChromiumAppBoundKeyCredential, Tag("chromium_app_bound_key")],
     Field(discriminator=Discriminator(get_credential_type)),
 ]

libs/common/common/state_helpers.py

@@ -111,7 +111,7 @@ def get_file_enriched(object_id: str) -> FileEnriched:
                     raise RuntimeError("Query returned no column descriptions")
 
                 columns = [desc[0] for desc in cur.description]
-                file_data = dict(zip(columns, result))
+                file_data = dict(zip(columns, result, strict=True))
 
                 # Transform data using shared helper
                 file_data = _transform_file_enriched_data(file_data)

libs/common/common/workflows/workflow_purger.py

@@ -318,7 +318,7 @@ async def _run_purge_cycle(self) -> dict:
                         # Collect successfully purged workflow IDs
                         purged_wf_ids = []
 
-                        for wf_id, result in zip(workflow_ids, purge_results):
+                        for wf_id, result in zip(workflow_ids, purge_results, strict=True):
                             if isinstance(result, Exception):
                                 logger.error(
                                     "Purge operation failed with exception",

libs/file_enrichment_modules/file_enrichment_modules/dotnet/analyzer.py

@@ -2,7 +2,6 @@
 import hashlib
 import json
 from pathlib import Path
-from typing import Union
 
 import dnfile
 from common.logger import get_logger
@@ -57,7 +56,7 @@ def process_resources(assembly):
     return resources
 
 
-def parse_dotnet_assembly(filename: Union[str, Path]) -> dict:
+def parse_dotnet_assembly(filename: str | Path) -> dict:
     """
     Parses a .NET assembly file and returns a dictionary of .NET specific data.
     """

libs/file_enrichment_modules/file_enrichment_modules/sqlite/analyzer.py

@@ -72,7 +72,9 @@ def format_sqlite_data(database_data: dict) -> str:
             continue
 
         # Show schema with column types
-        schema_with_types = [f"{col} ({type_})" for col, type_ in zip(data["schema"], data["column_types"])]
+        schema_with_types = [
+            f"{col} ({type_})" for col, type_ in zip(data["schema"], data["column_types"], strict=True)
+        ]
         output.append(f"Schema: {', '.join(schema_with_types)}")
 
         output.append("Data:")

projects/agents/agents/schemas.py

@@ -1,5 +1,5 @@
 from enum import StrEnum
-from typing import Any, Literal, Union
+from typing import Any, Literal
 
 from pydantic import BaseModel, Field, field_validator
 
@@ -27,7 +27,7 @@ class TriageRequest(BaseModel):
     finding_id: int = Field(..., description="Unique finding identifier")
     finding_name: str = Field(..., description="Name/type of the finding")
     category: str | None = Field(None, description="Finding category")
-    severity: Union[int, str] | None = Field(None, description="Finding severity level (0-10 or string)")
+    severity: int | str | None = Field(None, description="Finding severity level (0-10 or string)")
     object_id: str = Field(..., description="Object storage ID")
     origin_type: str | None = Field(None, description="Type of the finding's origin")
     origin_name: str | None = Field(None, description="Name of the finding's origin")

projects/cli/cli/cobaltstrike_connector/download_processor.py

@@ -5,7 +5,6 @@
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
-from typing import Union
 
 import plyvel
 from cli.cobaltstrike_connector.cobaltstrike_client import Beacon, CobaltStrikeClient, Download
@@ -92,7 +91,7 @@ def mark_processed(
 
     async def upload_file(
         self,
-        file_path: Union[Path, str],
+        file_path: Path | str,
         metadata: FileMetadata,
         delete_after: bool = False,
     ) -> tuple[bool, str | None, FileWithMetadataResponse | None]:

projects/cli/cli/config.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Annotated, Union
+from typing import Annotated
 from urllib.parse import urlparse
 
 import yaml
@@ -63,7 +63,7 @@ class NemesisConfig(BaseConfig):
 
 class MythicConfig(BaseConfig):
     url: StrictHttpUrl
-    credential: Union[PasswordCredential, TokenCredential]
+    credential: PasswordCredential | TokenCredential
 
     @field_validator("credential")
     @classmethod