mirror of
https://github.com/reconurge/flowsint.git
synced 2026-11-14 18:11:39 -06:00
dextmorgn
24e1e276c8
Each is a real defect, not style — flake8 just never ran before to catch them: - logger.py: bare `except:` around a queue drain swallowed everything including KeyboardInterrupt/SystemExit instead of just queue.Empty - dockertool.py: bare `except:` narrowed to Exception - permissions.py: duplicate `HTTPException` import, dead code - test_registry.py: `test_get_by_lower_key` was defined twice — the second definition silently shadowed the first, so only one ever ran - to_ports.py / test_investigation_repository.py: assigned-but-unused local variables, dead code - to_whois.py / to_infos.py: documented pyflakes false positive (OutputType is a class attr, resolves fine at def-time) — noqa'd with an explanation instead of restructuring working code - E712 (`== True`/`== False` → `is True`/`is False`) across 7 test files, F541 (f-strings with no placeholders) across 5 enrichers, E402 (import order) noqa'd in 2 files where the order is deliberate (env vars set before import; documented circular-import workaround)
73 lines
2.0 KiB
Python
73 lines
2.0 KiB
Python
from typing import List
|
|
|
|
from flowsint_core.core.enricher_base import Enricher
|
|
from flowsint_types import Phone
|
|
from flowsint_types.domain import Domain
|
|
from flowsint_types.ip import Ip
|
|
|
|
|
|
class ResolveEnricher(Enricher):
|
|
InputType = List[Domain]
|
|
OutputType = List[Ip]
|
|
|
|
@classmethod
|
|
def name(cls) -> str:
|
|
return "domain_to_ip"
|
|
|
|
@classmethod
|
|
def category(cls) -> str:
|
|
return "Domain"
|
|
|
|
@classmethod
|
|
def key(cls) -> str:
|
|
return "domain"
|
|
|
|
async def scan(self, data: InputType) -> OutputType:
|
|
return []
|
|
|
|
def postprocess(self, results: OutputType, original_input: InputType) -> OutputType:
|
|
return []
|
|
|
|
|
|
# Make types available at module level for easy access
|
|
InputType = ResolveEnricher.InputType
|
|
OutputType = ResolveEnricher.OutputType
|
|
|
|
enricher = ResolveEnricher("sketch_123", "scan_123")
|
|
|
|
|
|
def test_correct_preprocess():
|
|
inputs = [
|
|
Domain(domain="mydomain.com"),
|
|
{"domain": "blog.mydomain2.com"},
|
|
"mydomain3.com",
|
|
"notADomaiN",
|
|
]
|
|
preprocessed = enricher.preprocess(inputs)
|
|
assert len(preprocessed) == 3 # 3 valid domains
|
|
assert preprocessed[0].domain == "mydomain.com"
|
|
assert preprocessed[0].label == "mydomain.com"
|
|
assert preprocessed[0].root is True
|
|
|
|
assert preprocessed[1].domain == "blog.mydomain2.com"
|
|
assert preprocessed[1].label == "blog.mydomain2.com"
|
|
assert preprocessed[1].root is False
|
|
|
|
assert preprocessed[2].domain == "mydomain3.com"
|
|
assert preprocessed[2].label == "mydomain3.com"
|
|
assert preprocessed[2].root is True
|
|
|
|
|
|
def test_incorrect_preprocess():
|
|
inputs = [
|
|
Phone(number="+33634565423"),
|
|
{"name": "JohnDoe"},
|
|
"mydomain.com",
|
|
"notADomaiN",
|
|
]
|
|
preprocessed = enricher.preprocess(inputs)
|
|
assert len(preprocessed) == 1 # 1 valid domain
|
|
assert preprocessed[0].domain == "mydomain.com"
|
|
assert preprocessed[0].root is True
|
|
assert preprocessed[0].label == "mydomain.com"
|