fix: Replace 22 deprecated datetime.utcnow() calls with timezone-aware alternative

Replaced across 11 files: now uses datetime.now(tz=timezone.utc) per Python 3.12+
deprecation guidance. Removed redundant 'Z' suffixes where .isoformat() now
includes +00:00. Used .replace(tzinfo=None) for naive datetime arithmetic in
threat_intel_enricher.py KEV date comparison.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: devatsecure

scripts/feedback_collector.py

@@ -13,7 +13,7 @@
 
 import json
 import logging
-from datetime import datetime
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Literal, Optional
 
@@ -43,7 +43,7 @@ def record_feedback(
         feedback: Literal["tp", "fp"],
         reason: str,
         finding_details: Optional[dict[str, Any]] = None,
-        user: str = "user"
+        user: str = "user",
     ) -> bool:
         """
         Store feedback for future model improvement
@@ -69,7 +69,7 @@ def record_feedback(
                 "feedback_label": "true_positive" if feedback == "tp" else "false_positive",
                 "reason": reason,
                 "user": user,
-                "timestamp": datetime.utcnow().isoformat(),
+                "timestamp": datetime.now(tz=timezone.utc).isoformat(),
             }
 
             # Include finding details if provided
@@ -86,21 +86,15 @@ def record_feedback(
             with open(self.feedback_file, "a") as f:
                 f.write(json.dumps(feedback_entry) + "\n")
 
-            logger.info(
-                f"Recorded feedback: finding={finding_id}, "
-                f"feedback={feedback}, reason='{reason[:50]}...'"
-            )
+            logger.info(f"Recorded feedback: finding={finding_id}, feedback={feedback}, reason='{reason[:50]}...'")
 
             return True
 
         except Exception as e:
             logger.error(f"Failed to record feedback: {e}")
             return False
 
-    def get_all_feedback(
-        self,
-        feedback_type: Optional[Literal["tp", "fp"]] = None
-    ) -> list[dict[str, Any]]:
+    def get_all_feedback(self, feedback_type: Optional[Literal["tp", "fp"]] = None) -> list[dict[str, Any]]:
         """
         Retrieve all feedback entries
 
@@ -157,12 +151,7 @@ def get_feedback_by_id(self, finding_id: str) -> Optional[dict[str, Any]]:
 
         return None
 
-    def get_similar_findings(
-        self,
-        finding_type: str,
-        scanner: str,
-        limit: int = 5
-    ) -> list[dict[str, Any]]:
+    def get_similar_findings(self, finding_type: str, scanner: str, limit: int = 5) -> list[dict[str, Any]]:
         """
         Retrieve past feedback for similar findings (for few-shot prompting)
 
@@ -185,24 +174,15 @@ def get_similar_findings(
         similar = []
         for entry in all_feedback:
             finding = entry.get("finding", {})
-            if (finding.get("scanner") == scanner and
-                finding.get("finding_type") == finding_type):
+            if finding.get("scanner") == scanner and finding.get("finding_type") == finding_type:
                 similar.append(entry)
 
         # Sort by timestamp (most recent first)
-        similar.sort(
-            key=lambda x: x.get("timestamp", ""),
-            reverse=True
-        )
+        similar.sort(key=lambda x: x.get("timestamp", ""), reverse=True)
 
         return similar[:limit]
 
-    def generate_few_shot_examples(
-        self,
-        finding_type: str,
-        scanner: str,
-        max_examples: int = 3
-    ) -> str:
+    def generate_few_shot_examples(self, finding_type: str, scanner: str, max_examples: int = 3) -> str:
         """
         Generate few-shot prompt examples from historical feedback
 
@@ -229,9 +209,9 @@ def generate_few_shot_examples(
 
             example = f"""
 Example {i}:
-Finding Type: {finding.get('finding_type', 'unknown')}
-Scanner: {finding.get('scanner', 'unknown')}
-Description: {finding.get('description', '')[:150]}
+Finding Type: {finding.get("finding_type", "unknown")}
+Scanner: {finding.get("scanner", "unknown")}
+Description: {finding.get("description", "")[:150]}
 User Feedback: {feedback_label.upper()}
 Reason: {reason}
 """
@@ -344,19 +324,16 @@ def export_for_training(self, output_file: str) -> bool:
                         "messages": [
                             {
                                 "role": "system",
-                                "content": "You are a security analysis expert. Evaluate if findings are true positives or false positives."
+                                "content": "You are a security analysis expert. Evaluate if findings are true positives or false positives.",
                             },
                             {
                                 "role": "user",
                                 "content": f"Finding Type: {finding.get('finding_type')}\n"
-                                          f"Scanner: {finding.get('scanner')}\n"
-                                          f"Description: {finding.get('description')}\n"
-                                          f"Is this a true positive or false positive?"
+                                f"Scanner: {finding.get('scanner')}\n"
+                                f"Description: {finding.get('description')}\n"
+                                f"Is this a true positive or false positive?",
                             },
-                            {
-                                "role": "assistant",
-                                "content": f"{feedback_label.upper()}: {reason}"
-                            }
+                            {"role": "assistant", "content": f"{feedback_label.upper()}: {reason}"},
                         ]
                     }
 
@@ -392,25 +369,16 @@ def main():
     """CLI interface for feedback management"""
     import argparse
 
-    parser = argparse.ArgumentParser(
-        description="Manage user feedback on security findings"
-    )
-    parser.add_argument(
-        "--feedback-dir",
-        default=".argus/feedback",
-        help="Feedback directory path"
-    )
+    parser = argparse.ArgumentParser(description="Manage user feedback on security findings")
+    parser.add_argument("--feedback-dir", default=".argus/feedback", help="Feedback directory path")
 
     subparsers = parser.add_subparsers(dest="command", help="Command to execute")
 
     # Record feedback
     record_parser = subparsers.add_parser("record", help="Record feedback for a finding")
     record_parser.add_argument("finding_id", help="Finding ID")
     record_parser.add_argument(
-        "--mark",
-        choices=["tp", "fp"],
-        required=True,
-        help="Mark as true positive (tp) or false positive (fp)"
+        "--mark", choices=["tp", "fp"], required=True, help="Mark as true positive (tp) or false positive (fp)"
     )
     record_parser.add_argument("--reason", required=True, help="Reason for feedback")
 
@@ -423,11 +391,7 @@ def main():
 
     # List feedback
     list_parser = subparsers.add_parser("list", help="List all feedback")
-    list_parser.add_argument(
-        "--type",
-        choices=["tp", "fp"],
-        help="Filter by feedback type"
-    )
+    list_parser.add_argument("--type", choices=["tp", "fp"], help="Filter by feedback type")
 
     args = parser.parse_args()
 
@@ -438,11 +402,7 @@ def main():
     collector = FeedbackCollector(args.feedback_dir)
 
     if args.command == "record":
-        success = collector.record_feedback(
-            finding_id=args.finding_id,
-            feedback=args.mark,
-            reason=args.reason
-        )
+        success = collector.record_feedback(finding_id=args.finding_id, feedback=args.mark, reason=args.reason)
         if success:
             print(f"✅ Recorded feedback: {args.mark.upper()} for finding {args.finding_id}")
         else:
@@ -463,8 +423,9 @@ def main():
             print("\nBy Scanner:")
             for scanner, scanner_stats in stats["by_scanner"].items():
                 fp_rate = (scanner_stats["fp"] / scanner_stats["total"] * 100) if scanner_stats["total"] > 0 else 0
-                print(f"  {scanner:20s}: {scanner_stats['total']:3d} total, "
-                      f"{scanner_stats['fp']:3d} FP ({fp_rate:.0f}%)")
+                print(
+                    f"  {scanner:20s}: {scanner_stats['total']:3d} total, {scanner_stats['fp']:3d} FP ({fp_rate:.0f}%)"
+                )
 
         if stats.get("recent_feedback"):
             print("\nRecent Feedback:")
@@ -484,7 +445,7 @@ def main():
     elif args.command == "list":
         feedback_list = collector.get_all_feedback(feedback_type=args.type)
 
-        print(f"\n{'='*80}")
+        print(f"\n{'=' * 80}")
         print(f"FEEDBACK LIST ({len(feedback_list)} entries)")
         print("=" * 80)
 

scripts/orchestrator/llm_manager.py

@@ -146,24 +146,28 @@ def _apply_retry_strategy(self):
                 provider=self.config.get("ai_provider", ""),
             )(self.call_llm_api)
             logger.debug(
-                "Smart retry enabled (max_attempts=%d)", max_attempts,
+                "Smart retry enabled (max_attempts=%d)",
+                max_attempts,
             )
         else:
             # Legacy tenacity retry for backward compatibility
             self.call_llm_api = retry(
                 stop=stop_after_attempt(max_attempts),
                 wait=wait_exponential(multiplier=1, min=4, max=10),
-                retry=retry_if_exception_type((
-                    ConnectionError,
-                    TimeoutError,
-                    OSError,
-                    Exception,
-                )),
+                retry=retry_if_exception_type(
+                    (
+                        ConnectionError,
+                        TimeoutError,
+                        OSError,
+                        Exception,
+                    )
+                ),
                 before_sleep=before_sleep_log(logger, logging.WARNING),
                 reraise=True,
             )(self.call_llm_api)
             logger.debug(
-                "Legacy tenacity retry enabled (max_attempts=%d)", max_attempts,
+                "Legacy tenacity retry enabled (max_attempts=%d)",
+                max_attempts,
             )
 
     def detect_provider(self) -> str:
@@ -190,7 +194,9 @@ def detect_provider(self) -> str:
             return "claude-cli"
         else:
             logger.warning("No AI provider configured")
-            logger.info("Set one of: ANTHROPIC_API_KEY, OPENAI_API_KEY, OLLAMA_ENDPOINT, or use --ai-provider claude-cli")
+            logger.info(
+                "Set one of: ANTHROPIC_API_KEY, OPENAI_API_KEY, OLLAMA_ENDPOINT, or use --ai-provider claude-cli"
+            )
             return None
 
     def initialize(self, provider: str = None) -> bool:
@@ -271,7 +277,9 @@ def _get_client(self, provider: str):
                 endpoint = self.config.get("ollama_endpoint", "http://localhost:11434")
                 # Sanitize endpoint URL for logging
                 safe_endpoint = (
-                    str(endpoint).split("@")[-1] if "@" in str(endpoint) else str(endpoint).split("//")[-1].split("/")[0]
+                    str(endpoint).split("@")[-1]
+                    if "@" in str(endpoint)
+                    else str(endpoint).split("//")[-1].split("/")[0]
                 )
                 logger.info(f"Using Ollama endpoint: {safe_endpoint}")
                 return OpenAI(base_url=f"{endpoint}/v1", api_key="ollama"), "ollama"
@@ -347,9 +355,7 @@ def _get_working_model_with_fallback(self, client, initial_model: str) -> str:
                 # Quick test with minimal tokens
                 safe_model_name = str(model_id).split("/")[-1] if model_id else "unknown"
                 logger.debug(f"Testing model: {safe_model_name}")
-                client.messages.create(
-                    model=model_id, max_tokens=10, messages=[{"role": "user", "content": "test"}]
-                )
+                client.messages.create(model=model_id, max_tokens=10, messages=[{"role": "user", "content": "test"}])
                 logger.info(f"Found working model: {safe_model_name}")
                 return model_id
             except Exception as e:
@@ -415,12 +421,7 @@ def calculate_actual_cost(input_tokens: int, output_tokens: int, provider: str)
 
         return input_cost + output_cost
 
-    def generate_few_shot_examples(
-        self,
-        finding_type: str,
-        scanner: str,
-        max_examples: int = 3
-    ) -> str:
+    def generate_few_shot_examples(self, finding_type: str, scanner: str, max_examples: int = 3) -> str:
         """
         Generate few-shot examples from historical feedback
 
@@ -437,9 +438,7 @@ def generate_few_shot_examples(
 
         try:
             return self.feedback_collector.generate_few_shot_examples(
-                finding_type=finding_type,
-                scanner=scanner,
-                max_examples=max_examples
+                finding_type=finding_type, scanner=scanner, max_examples=max_examples
             )
         except Exception as e:
             logger.debug(f"Could not generate few-shot examples: {e}")
@@ -453,7 +452,7 @@ def log_ai_decision(
         decision: str,
         reasoning: str,
         confidence: float,
-        noise_score: float = 0.0
+        noise_score: float = 0.0,
     ) -> bool:
         """
         Log AI triage decision for analysis
@@ -474,7 +473,7 @@ def log_ai_decision(
             return False
 
         try:
-            from datetime import datetime
+            from datetime import datetime, timezone
 
             decision_entry = {
                 "finding_id": finding_id,
@@ -485,7 +484,7 @@ def log_ai_decision(
                 "confidence": confidence,
                 "noise_score": noise_score,
                 "model": self.model,
-                "timestamp": datetime.utcnow().isoformat(),
+                "timestamp": datetime.now(tz=timezone.utc).isoformat(),
             }
 
             return self.cache_manager.log_decision(decision_entry)
@@ -500,7 +499,7 @@ def call_llm_api(
         max_tokens: int,
         circuit_breaker: "CostCircuitBreaker" = None,
         operation: str = "LLM call",
-        few_shot_prefix: str = ""
+        few_shot_prefix: str = "",
     ) -> tuple:
         """Call LLM API with retry logic, cost enforcement, and few-shot learning
 
@@ -563,11 +562,14 @@ def call_llm_api(
                     [
                         self.client,  # path to claude binary
                         "--print",
-                        "--model", self.model,
-                        "--output-format", "json",
+                        "--model",
+                        self.model,
+                        "--output-format",
+                        "json",
                         "--no-session-persistence",
                         "--dangerously-skip-permissions",
-                        "--max-turns", "1",
+                        "--max-turns",
+                        "1",
                     ],
                     input=full_prompt,
                     capture_output=True,
@@ -635,6 +637,7 @@ def generate(self, user_prompt: str, system_prompt: str = "", max_tokens: int =
 
 class _Usage:
     """Minimal usage object matching Anthropic/OpenAI response.usage."""
+
     __slots__ = ("input_tokens", "output_tokens")
 
     def __init__(self, input_tokens: int, output_tokens: int):
@@ -644,6 +647,7 @@ def __init__(self, input_tokens: int, output_tokens: int):
 
 class LLMResponse:
     """Lightweight response wrapper compatible with IRISAnalyzer._parse_llm_response."""
+
     __slots__ = ("content", "usage")
 
     def __init__(self, text: str, input_tokens: int = 0, output_tokens: int = 0):
@@ -741,7 +745,9 @@ def calculate_actual_cost(input_tokens: int, output_tokens: int, provider: str)
     return LLMManager.calculate_actual_cost(input_tokens, output_tokens, provider)
 
 
-def call_llm_api(client, provider: str, model: str, prompt: str, max_tokens: int, circuit_breaker=None, operation: str = "LLM call") -> tuple:
+def call_llm_api(
+    client, provider: str, model: str, prompt: str, max_tokens: int, circuit_breaker=None, operation: str = "LLM call"
+) -> tuple:
     """Call LLM API with retry logic and cost enforcement
 
     Args: