fix: resolve Ruff 0.16.0 lint/format violations

.github/workflows/lint_format_checker.yaml

@@ -19,8 +19,8 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
       - name: Checking code formatting.
-        uses: chartboost/ruff-action@v1
+        uses: astral-sh/ruff-action@v3
         with:
           args: format --check
       - name: Running linter.
-        uses: chartboost/ruff-action@v1
+        uses: astral-sh/ruff-action@v3

agent/mcp_server/server.py

@@ -7,15 +7,14 @@
 import time
 
 import click
-import google.cloud.logging
-from google.oauth2 import service_account
 import fastmcp
+import google.cloud.logging
 from fastmcp import tools as fastmcp_tools
+from google.oauth2 import service_account
 from rich import logging as rich_logging
 
 from agent.mcp_server import tools
 
-
 logging.basicConfig(
     format="%(message)s",
     datefmt="[%X]",

agent/whatweb_agent.py

@@ -9,11 +9,9 @@
 import re
 import subprocess
 import tempfile
-from typing import List, Optional, Dict, Any
+from typing import Any
 from urllib import parse
 
-from agent.mcp_server import mcp_runner
-
 from ostorlab.agent import agent
 from ostorlab.agent import definitions as agent_definitions
 from ostorlab.agent.kb import kb
@@ -27,6 +25,7 @@
 from rich import logging as rich_logging
 
 from agent import definitions
+from agent.mcp_server import mcp_runner
 
 logging.basicConfig(
     format="%(message)s",
@@ -62,8 +61,8 @@ class DomainTarget(BaseTarget):
     """Domain target."""
 
     name: str
-    schema: Optional[str] = None
-    port: Optional[int] = None
+    schema: str | None = None
+    port: int | None = None
 
     @property
     def target(self) -> str:
@@ -86,8 +85,8 @@ class IPTarget(BaseTarget):
 
     name: str
     version: int
-    schema: Optional[str] = None
-    port: Optional[int] = None
+    schema: str | None = None
+    port: int | None = None
 
     @property
     def target(self) -> str:
@@ -117,7 +116,7 @@ def __init__(
         agent.Agent.__init__(self, agent_definition, agent_settings)
         vuln_mixin.AgentReportVulnMixin.__init__(self)
         persist_mixin.AgentPersistMixin.__init__(self, agent_settings)
-        self._scope_domain_regex: Optional[str] = self.args.get("scope_domain_regex")
+        self._scope_domain_regex: str | None = self.args.get("scope_domain_regex")
         self._should_start_mcp_server: bool = self.args.get(
             "should_start_mcp_server", False
         )
@@ -162,9 +161,9 @@ def process(self, message: msg.Message) -> None:
             except subprocess.CalledProcessError as e:
                 logger.error("Error scanning target `%s`: %s", target, e)
 
-    def _prepare_targets(self, message: msg.Message) -> List[IPTarget | DomainTarget]:
+    def _prepare_targets(self, message: msg.Message) -> list[IPTarget | DomainTarget]:
         """Returns a list of target objects to be scanned."""
-        targets: List[DomainTarget | IPTarget] = []
+        targets: list[DomainTarget | IPTarget] = []
         domain_targets = self._prepare_domain_targets(message)
         ip_targets = self._prepare_ip_targets(message)
         targets.extend(domain_targets)
@@ -187,9 +186,9 @@ def _get_schema(self, message: msg.Message) -> str:
         else:
             return str(self.args["schema"])
 
-    def _prepare_domain_targets(self, message: msg.Message) -> List[DomainTarget]:
+    def _prepare_domain_targets(self, message: msg.Message) -> list[DomainTarget]:
         """Returns a list of domain targets to be scanned."""
-        targets: List[DomainTarget] = []
+        targets: list[DomainTarget] = []
         if message.data.get("url") is not None:
             url_target = self._get_target_from_url(message.data["url"])
             if url_target is not None:
@@ -204,9 +203,9 @@ def _prepare_domain_targets(self, message: msg.Message) -> List[DomainTarget]:
             targets.append(domain_target)
         return targets
 
-    def _prepare_ip_targets(self, message: msg.Message) -> List[IPTarget]:
+    def _prepare_ip_targets(self, message: msg.Message) -> list[IPTarget]:
         """Returns a list of ip targets to be scanned."""
-        targets: List[IPTarget] = []
+        targets: list[IPTarget] = []
         host = message.data.get("host")
         mask = message.data.get("mask")
         if host is None:
@@ -426,8 +425,8 @@ def _prepare_vulnerable_target_data(
     def _send_detected_fingerprints(
         self,
         target: DomainTarget | IPTarget,
-        library_name: Optional[str] = None,
-        versions: Optional[List[Optional[str]]] = None,
+        library_name: str | None = None,
+        versions: list[str | None] | None = None,
     ) -> None:
         """Emits the identified fingerprints.
 
@@ -476,7 +475,7 @@ def _send_detected_fingerprints(
                         targeted_by_ransomware=False,
                         targeted_by_nation_state=False,
                     ),
-                    technical_detail=f"Found fingerprint `{library_name}`, version `{str(version)}`, "
+                    technical_detail=f"Found fingerprint `{library_name}`, version `{version!s}`, "
                     f"of type `{fingerprint_type}` in target `{target.name}`",
                     risk_rating=vuln_mixin.RiskRating.INFO,
                     vulnerability_location=vulnerable_target_data,
@@ -514,12 +513,12 @@ def _send_detected_fingerprints(
     def _get_msg_data(
         self,
         target: DomainTarget | IPTarget,
-        library_name: Optional[str] = None,
-        version: Optional[str] = None,
-        fingerprint_type: Optional[str] = None,
-    ) -> Dict[str, Any]:
+        library_name: str | None = None,
+        version: str | None = None,
+        fingerprint_type: str | None = None,
+    ) -> dict[str, Any]:
         """Prepare  data of the library proto message to be emitted."""
-        msg_data: Dict[str, Any] = {}
+        msg_data: dict[str, Any] = {}
         if target.name is not None:
             if isinstance(target, DomainTarget):
                 msg_data["name"] = target.name
@@ -537,7 +536,7 @@ def _get_msg_data(
         if version is not None:
             msg_data["library_version"] = str(version)
             detail = (
-                f"Found fingerprint `{library_name}`, version `{str(version)}`, of type"
+                f"Found fingerprint `{library_name}`, version `{version!s}`, of type"
             )
             detail = f"{detail} `{fingerprint_type}` in target `{target.name}`"
             msg_data["detail"] = detail

tests/conftest.py

@@ -1,15 +1,14 @@
 """Pytest fixture for the whatweb agent."""
 
-import pytest
 import json
 import pathlib
-from typing import Dict, Union
 import random
 
+import pytest
 from ostorlab.agent import definitions as agent_definitions
+from ostorlab.agent.message import message as m
 from ostorlab.runtimes import definitions as runtime_definitions
 from ostorlab.utils import definitions
-from ostorlab.agent.message import message as m
 
 from agent import whatweb_agent
 
@@ -100,7 +99,7 @@ def ip_tcp_message() -> m.Message:
 
 @pytest.fixture(scope="function")
 def whatweb_test_agent(
-    agent_persist_mock: Dict[Union[str, bytes], Union[str, bytes]],
+    agent_persist_mock: dict[str | bytes, str | bytes],
 ) -> whatweb_agent.AgentWhatWeb:
     """WhatWeb Agent fixture for testing purposes."""
     del agent_persist_mock
@@ -126,7 +125,7 @@ def whatweb_test_agent(
 
 @pytest.fixture(scope="function")
 def whatweb_agent_with_scope_arg(
-    agent_persist_mock: Dict[Union[str, bytes], Union[str, bytes]],
+    agent_persist_mock: dict[str | bytes, str | bytes],
 ) -> whatweb_agent.AgentWhatWeb:
     """WhatWeb Agent fixture for testing purposes."""
     del agent_persist_mock
@@ -214,7 +213,7 @@ def scan_message_ipv_with_incorrect_version() -> m.Message:
 
 @pytest.fixture(scope="function")
 def whatweb_agent_with_mcp_server(
-    agent_persist_mock: Dict[Union[str, bytes], Union[str, bytes]],
+    agent_persist_mock: dict[str | bytes, str | bytes],
 ) -> whatweb_agent.AgentWhatWeb:
     """WhatWeb Agent fixture with MCP server enabled for testing purposes."""
     del agent_persist_mock

tests/fingerprint_mcp_server_test.py

@@ -6,8 +6,7 @@
 import pytest
 from pytest_mock import plugin
 
-from agent import definitions
-from agent import whatweb_utils
+from agent import definitions, whatweb_utils
 from agent.mcp_server import tools
 
 TESTS_DIR = pathlib.Path(__file__).parent

tests/whatweb_test.py

@@ -700,7 +700,7 @@ def testWhatWebAgent_withIPv4AndMaskButNoVersion_shouldHandleVersionCorrectly(
     assert len(calls) == len(expected_ips)
 
     for call, expected_ip in zip(calls, expected_ips):
-        args, kwargs = call
+        args, _kwargs = call
         command = args[0]
         assert any(expected_ip in arg for arg in command), (
             f"Expected IP {expected_ip} not found in command {command}"