| 1 | """Enricher registry - the plugin system. New OSINT sources register here and are |
| 2 | immediately available via the API without touching core code.""" |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from app.services.enrichment.base import BaseEnricher, EnrichResult |
| 6 | from app.services.enrichment.plugins import ( |
| 7 | CrtShEnricher, DnsEnricher, GithubEnricher, IpAsnEnricher, |
| 8 | RdapEnricher, RobotsEnricher, SecurityTxtEnricher, TlsCertEnricher, |
| 9 | ) |
| 10 | |
| 11 | _REGISTRY: dict[str, BaseEnricher] = { |
| 12 | e.name: e for e in [ |
| 13 | DnsEnricher(), RdapEnricher(), CrtShEnricher(), IpAsnEnricher(), |
| 14 | SecurityTxtEnricher(), RobotsEnricher(), GithubEnricher(), TlsCertEnricher(), |
| 15 | ] |
| 16 | } |
| 17 | |
| 18 | |
| 19 | def list_enrichers() -> list[dict]: |
| 20 | return [{"name": e.name, "kind": e.kind, "requires_network": e.requires_network} |
| 21 | for e in _REGISTRY.values()] |
| 22 | |
| 23 | |
| 24 | def get_enricher(name: str) -> BaseEnricher | None: |
| 25 | return _REGISTRY.get(name) |
| 26 | |
| 27 | |
| 28 | def run_enricher(name: str, target: str) -> EnrichResult: |
| 29 | enricher = _REGISTRY.get(name) |
| 30 | if not enricher: |
| 31 | return EnrichResult(enricher=name, target=target, error=f"unknown enricher '{name}'", confidence=0.0) |
| 32 | return enricher.run(target) |
| 33 | |
| 34 | |
| 35 | __all__ = ["list_enrichers", "get_enricher", "run_enricher", "EnrichResult"] |