ratchet(types): promote invalid-method-override + invalid-return-type (plugins)

core/schemas/task.py

@@ -246,7 +246,7 @@ def analysis_done(self, observable_id: str):
         observable.last_analysis[self.name] = now()
         observable.save()
 
-    def each(self, observable: Observable) -> Observable:
+    def each(self, observable: Observable, /) -> Observable:
         """Analyzes a single observable.
 
         Args:
@@ -280,7 +280,7 @@ def run(self, params: dict):
             return
         self.each(results[0])
 
-    def each(self, observable: Observable) -> None:
+    def each(self, observable: Observable, /) -> None:
         """Analyzes a single observable.
 
         Args:

plugins/analytics/public/circl_passive_ssl.py

@@ -1,3 +1,5 @@
+from typing import cast
+
 import requests
 from OpenSSL.crypto import (
     FILETYPE_ASN1,
@@ -9,7 +11,7 @@
 from core import taskmanager
 from core.config.config import yeti_config
 from core.schemas import task
-from core.schemas.observable import ObservableType
+from core.schemas.observable import Observable, ObservableType
 from core.schemas.observables import certificate, ipv4
 
 
@@ -68,7 +70,8 @@ class CirclPassiveSSLSearchIP(task.AnalyticsTask, CirclPassiveSSLApi):
 
     acts_on: list[ObservableType] = [ObservableType.ipv4]
 
-    def each(self, ip: ipv4.IPv4):
+    def each(self, observable: Observable):
+        ip = cast("ipv4.IPv4", observable)
         ip_search = CirclPassiveSSLApi.search_ip(ip)
         if ip_search:
             for ip_addr, ip_details in ip_search.items():

plugins/analytics/public/dockerhub.py

@@ -21,7 +21,7 @@ def _make_request(endpoint) -> dict:
         return {}
 
     @staticmethod
-    def _iter_endpoint_pages(endpoint) -> dict:
+    def _iter_endpoint_pages(endpoint) -> Iterator:
         data = DockerHubApi._make_request(endpoint)
         while data:
             for result in data.get("results", []):

plugins/analytics/public/macaddress_io.py

@@ -2,6 +2,7 @@
 
 import json
 from datetime import datetime
+from typing import cast
 
 from maclookup import ApiClient
 from maclookup import exceptions as maclookup_exceptions
@@ -10,7 +11,7 @@
 from core.config.config import yeti_config
 from core.schemas import task
 from core.schemas.entity import Company
-from core.schemas.observable import ObservableType
+from core.schemas.observable import Observable, ObservableType
 from core.schemas.observables.mac_address import MacAddress
 
 
@@ -62,7 +63,8 @@ class MacAddressIo(task.AnalyticsTask, MacAddressIoApi):
 
     acts_on: list[ObservableType] = [ObservableType.mac_address]
 
-    def each(self, mac_address: MacAddress):
+    def each(self, observable: Observable):
+        mac_address = cast("MacAddress", observable)
         lookup_results = MacAddressIoApi.get(mac_address.value)
 
         vendor = None

plugins/analytics/public/network_whois.py

@@ -1,9 +1,11 @@
+from typing import cast
+
 from ipwhois import IPWhois
 
 from core import taskmanager
 from core.schemas import task
 from core.schemas.entity import Company
-from core.schemas.observable import ObservableType
+from core.schemas.observable import Observable, ObservableType
 from core.schemas.observables import email, ipv4
 
 
@@ -16,7 +18,8 @@ class NetworkWhois(task.AnalyticsTask):
 
     acts_on: list[ObservableType] = [ObservableType.ipv4]
 
-    def each(self, ip: ipv4.IPv4):
+    def each(self, observable: Observable):
+        ip = cast("ipv4.IPv4", observable)
         r = IPWhois(ip.value)
         result = r.lookup_whois()
 

plugins/analytics/public/shodan_api.py

@@ -1,4 +1,5 @@
 import logging
+from typing import cast
 
 import shodan
 
@@ -36,7 +37,8 @@ class ShodanQuery(task.OneShotTask, ShodanApi):
 
     acts_on: list[ObservableType] = [ObservableType.ipv4]
 
-    def each(self, ip: ipv4.IPv4) -> Observable:
+    def each(self, observable: Observable) -> None:
+        ip = cast("ipv4.IPv4", observable)
         result = ShodanApi.fetch(ip)
         logging.debug(result)
 
@@ -60,7 +62,7 @@ def each(self, ip: ipv4.IPv4) -> Observable:
             logging.debug(result["isp"])
             o_isp = Company(name=result["isp"]).save()
             ip.link_to(o_isp, "hosting", "Shodan Query")
-        return ip
+        return
 
 
 taskmanager.TaskManager.register_task(ShodanQuery)

plugins/feeds/public/et_open.py

@@ -2,6 +2,7 @@
 import logging
 from datetime import timedelta, timezone
 from io import StringIO
+from typing import cast
 
 from idstools import rule
 
@@ -68,7 +69,7 @@ def _extract_cve(self, meta: str) -> entity.Vulnerability:
         ind_cve = entity.Vulnerability.find(name=cve)
         if not ind_cve:
             ind_cve = entity.Vulnerability(name=cve).save()
-        return ind_cve
+        return cast("entity.Vulnerability", ind_cve)
 
     def _extract_malware_family(self, meta: str):
         _, malware_family = meta.split(" ")
@@ -92,7 +93,7 @@ def _extract_mitre_attack(self, meta: str) -> entity.AttackPattern | None:
             aliases=[("text", mitre_id)],
         )
         if nb_ent != 0:
-            return ind_mitre_attack[0]
+            return cast("entity.AttackPattern", ind_mitre_attack[0])
 
     def _filter_rule(self, metadata: list[str]):
         """

plugins/feeds/public/miningpoolstats.py

@@ -83,7 +83,7 @@ def _extract_pool_urls(self, coin_name: str) -> Generator[dict, None, None]:
             yield data
         return
 
-    def _get_coin_names(self) -> list:
+    def _get_coin_names(self) -> set | None:
         """
         Return available coin names in two steps. At first, fetch the main page to extract the
         timestamped endpoint. Then, add to a set the coin name from the page key.

pyproject.toml

@@ -114,8 +114,6 @@ include = ["plugins"]
 [tool.ty.overrides.rules]
 invalid-argument-type = "warn"      # 41
 unresolved-attribute = "warn"       # 32
-invalid-method-override = "warn"    # 12
-invalid-return-type = "warn"        # 7
 deprecated = "warn"                 # 5
 # unused-ignore-comment stays warn: the only instances are the boto3/tldextract
 # import ignores in core (s3.py/utils.py) — used by the dev-only main job but