40 lines
1.4 KiB
Python
40 lines
1.4 KiB
Python
"""Coordinate optional enrichment and progress callbacks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Callable, Optional, Sequence
|
|
|
|
from ..artifacts import HashArtifact, UrlArtifact
|
|
from .hash_crack import HashCrack, crack_hash
|
|
from .url_probe import UrlProbe, probe_url
|
|
|
|
# progress(done, total, item) -> None
|
|
ProgressCb = Optional[Callable[[int, int, object], None]]
|
|
|
|
|
|
class EnrichmentEngine:
|
|
"""Run hash and endpoint enrichment for artifact lists."""
|
|
|
|
def crack_hashes(self, artifacts: Sequence[HashArtifact], *,
|
|
progress: ProgressCb = None, **crack_kwargs) -> list[HashCrack]:
|
|
results: list[HashCrack] = []
|
|
total = len(artifacts)
|
|
for i, art in enumerate(artifacts, 1):
|
|
res = crack_hash(art, **crack_kwargs)
|
|
results.append(res)
|
|
if progress:
|
|
progress(i, total, res)
|
|
return results
|
|
|
|
def probe_urls(self, artifacts: Sequence[UrlArtifact], *,
|
|
progress: ProgressCb = None, active: bool = False,
|
|
**probe_kwargs) -> list[UrlProbe]:
|
|
results: list[UrlProbe] = []
|
|
total = len(artifacts)
|
|
for i, art in enumerate(artifacts, 1):
|
|
res = probe_url(art, active=active, **probe_kwargs)
|
|
results.append(res)
|
|
if progress:
|
|
progress(i, total, res)
|
|
return results
|