diff --git a/core/schemas/task.py b/core/schemas/task.py index 396868357..c009673a6 100644 --- a/core/schemas/task.py +++ b/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: diff --git a/plugins/analytics/public/circl_passive_ssl.py b/plugins/analytics/public/circl_passive_ssl.py index c75abce3e..7a8335b49 100644 --- a/plugins/analytics/public/circl_passive_ssl.py +++ b/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(): diff --git a/plugins/analytics/public/dockerhub.py b/plugins/analytics/public/dockerhub.py index 0b65b72e3..944c3850c 100644 --- a/plugins/analytics/public/dockerhub.py +++ b/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", []): diff --git a/plugins/analytics/public/macaddress_io.py b/plugins/analytics/public/macaddress_io.py index 81137c2c2..b172e68ec 100644 --- a/plugins/analytics/public/macaddress_io.py +++ b/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 diff --git a/plugins/analytics/public/network_whois.py b/plugins/analytics/public/network_whois.py index 265f82dc9..1723dcbbf 100644 --- a/plugins/analytics/public/network_whois.py +++ b/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() diff --git a/plugins/analytics/public/shodan_api.py b/plugins/analytics/public/shodan_api.py index 56e6ef8a9..4e48b1027 100644 --- a/plugins/analytics/public/shodan_api.py +++ b/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) diff --git a/plugins/feeds/public/et_open.py b/plugins/feeds/public/et_open.py index 86ed2e5f8..5022c476d 100644 --- a/plugins/feeds/public/et_open.py +++ b/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]): """ diff --git a/plugins/feeds/public/miningpoolstats.py b/plugins/feeds/public/miningpoolstats.py index b46fe40d8..c593c1d85 100644 --- a/plugins/feeds/public/miningpoolstats.py +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index bbdf36b6f..2460413f4 100644 --- a/pyproject.toml +++ b/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