implement core foundation: entity models, validators, http client, database
This commit is contained in:
parent
40a463ba03
commit
d1561f1120
11 changed files with 624 additions and 0 deletions
3
eirescope/__init__.py
Normal file
3
eirescope/__init__.py
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
"""EireScope — Open-source OSINT Investigation Dashboard."""
|
||||
__version__ = "0.1.0"
|
||||
__author__ = "EireScope Contributors"
|
||||
0
eirescope/core/__init__.py
Normal file
0
eirescope/core/__init__.py
Normal file
153
eirescope/core/entity.py
Normal file
153
eirescope/core/entity.py
Normal file
|
|
@ -0,0 +1,153 @@
|
|||
"""Core data models for EireScope investigations."""
|
||||
import uuid
|
||||
from enum import Enum
|
||||
from datetime import datetime
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Dict, Optional, Any
|
||||
|
||||
|
||||
class EntityType(Enum):
|
||||
"""All searchable/discoverable entity types."""
|
||||
EMAIL = "email"
|
||||
USERNAME = "username"
|
||||
PHONE = "phone"
|
||||
IP_ADDRESS = "ip_address"
|
||||
DOMAIN = "domain"
|
||||
COMPANY = "company"
|
||||
PERSON = "person"
|
||||
SOCIAL_PROFILE = "social_profile"
|
||||
URL = "url"
|
||||
HASH = "hash"
|
||||
BREACH = "breach"
|
||||
DNS_RECORD = "dns_record"
|
||||
WHOIS_INFO = "whois_info"
|
||||
GEO_LOCATION = "geo_location"
|
||||
CARRIER_INFO = "carrier_info"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Entity:
|
||||
"""Represents a single OSINT artifact discovered during investigation."""
|
||||
entity_type: EntityType
|
||||
value: str
|
||||
source_module: str = ""
|
||||
confidence: float = 1.0
|
||||
metadata: Dict[str, Any] = field(default_factory=dict)
|
||||
id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
||||
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
|
||||
|
||||
def to_dict(self) -> Dict:
|
||||
return {
|
||||
"id": self.id,
|
||||
"entity_type": self.entity_type.value,
|
||||
"value": self.value,
|
||||
"source_module": self.source_module,
|
||||
"confidence": self.confidence,
|
||||
"metadata": self.metadata,
|
||||
"created_at": self.created_at,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict) -> "Entity":
|
||||
return cls(
|
||||
id=data.get("id", str(uuid.uuid4())),
|
||||
entity_type=EntityType(data["entity_type"]),
|
||||
value=data["value"],
|
||||
source_module=data.get("source_module", ""),
|
||||
confidence=data.get("confidence", 1.0),
|
||||
metadata=data.get("metadata", {}),
|
||||
created_at=data.get("created_at", datetime.utcnow().isoformat()),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class EntityRelationship:
|
||||
"""Links two entities with a typed relationship."""
|
||||
source_entity_id: str
|
||||
target_entity_id: str
|
||||
relationship_type: str
|
||||
confidence: float = 1.0
|
||||
evidence: Dict[str, Any] = field(default_factory=dict)
|
||||
id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
||||
|
||||
def to_dict(self) -> Dict:
|
||||
return {
|
||||
"id": self.id,
|
||||
"source_entity_id": self.source_entity_id,
|
||||
"target_entity_id": self.target_entity_id,
|
||||
"relationship_type": self.relationship_type,
|
||||
"confidence": self.confidence,
|
||||
"evidence": self.evidence,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Investigation:
|
||||
"""A single OSINT investigation session."""
|
||||
initial_query: str
|
||||
initial_type: EntityType
|
||||
entities: List[Entity] = field(default_factory=list)
|
||||
relationships: List[EntityRelationship] = field(default_factory=list)
|
||||
modules_run: List[str] = field(default_factory=list)
|
||||
status: str = "pending" # pending, running, completed, failed
|
||||
notes: str = ""
|
||||
id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
||||
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
|
||||
completed_at: Optional[str] = None
|
||||
|
||||
def add_entity(self, entity: Entity) -> Entity:
|
||||
"""Add entity, deduplicating by type+value."""
|
||||
for existing in self.entities:
|
||||
if existing.entity_type == entity.entity_type and existing.value == entity.value:
|
||||
existing.metadata.update(entity.metadata)
|
||||
return existing
|
||||
self.entities.append(entity)
|
||||
return entity
|
||||
|
||||
def add_relationship(self, source_id: str, target_id: str,
|
||||
rel_type: str, confidence: float = 1.0,
|
||||
evidence: Dict = None) -> EntityRelationship:
|
||||
"""Add a relationship between two entities."""
|
||||
rel = EntityRelationship(
|
||||
source_entity_id=source_id,
|
||||
target_entity_id=target_id,
|
||||
relationship_type=rel_type,
|
||||
confidence=confidence,
|
||||
evidence=evidence or {},
|
||||
)
|
||||
self.relationships.append(rel)
|
||||
return rel
|
||||
|
||||
def complete(self):
|
||||
self.status = "completed"
|
||||
self.completed_at = datetime.utcnow().isoformat()
|
||||
|
||||
def fail(self, reason: str = ""):
|
||||
self.status = "failed"
|
||||
self.completed_at = datetime.utcnow().isoformat()
|
||||
self.notes = reason
|
||||
|
||||
def to_dict(self) -> Dict:
|
||||
return {
|
||||
"id": self.id,
|
||||
"initial_query": self.initial_query,
|
||||
"initial_type": self.initial_type.value,
|
||||
"entities": [e.to_dict() for e in self.entities],
|
||||
"relationships": [r.to_dict() for r in self.relationships],
|
||||
"modules_run": self.modules_run,
|
||||
"status": self.status,
|
||||
"notes": self.notes,
|
||||
"created_at": self.created_at,
|
||||
"completed_at": self.completed_at,
|
||||
"entity_count": len(self.entities),
|
||||
"relationship_count": len(self.relationships),
|
||||
}
|
||||
|
||||
def get_entities_by_type(self, entity_type: EntityType) -> List[Entity]:
|
||||
return [e for e in self.entities if e.entity_type == entity_type]
|
||||
|
||||
def get_entity_by_id(self, entity_id: str) -> Optional[Entity]:
|
||||
for e in self.entities:
|
||||
if e.id == entity_id:
|
||||
return e
|
||||
return None
|
||||
0
eirescope/db/__init__.py
Normal file
0
eirescope/db/__init__.py
Normal file
182
eirescope/db/database.py
Normal file
182
eirescope/db/database.py
Normal file
|
|
@ -0,0 +1,182 @@
|
|||
"""SQLite database for persisting EireScope investigations."""
|
||||
import os
|
||||
import json
|
||||
import sqlite3
|
||||
import logging
|
||||
from typing import List, Optional, Dict
|
||||
from datetime import datetime
|
||||
from eirescope.core.entity import Entity, EntityType, EntityRelationship, Investigation
|
||||
|
||||
logger = logging.getLogger("eirescope.db")
|
||||
|
||||
CREATE_TABLES_SQL = """
|
||||
CREATE TABLE IF NOT EXISTS investigations (
|
||||
id TEXT PRIMARY KEY,
|
||||
initial_query TEXT NOT NULL,
|
||||
initial_type TEXT NOT NULL,
|
||||
status TEXT DEFAULT 'pending',
|
||||
notes TEXT DEFAULT '',
|
||||
modules_run TEXT DEFAULT '[]',
|
||||
created_at TEXT NOT NULL,
|
||||
completed_at TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS entities (
|
||||
id TEXT PRIMARY KEY,
|
||||
investigation_id TEXT NOT NULL,
|
||||
entity_type TEXT NOT NULL,
|
||||
value TEXT NOT NULL,
|
||||
source_module TEXT DEFAULT '',
|
||||
confidence REAL DEFAULT 1.0,
|
||||
metadata TEXT DEFAULT '{}',
|
||||
created_at TEXT NOT NULL,
|
||||
FOREIGN KEY (investigation_id) REFERENCES investigations(id)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS relationships (
|
||||
id TEXT PRIMARY KEY,
|
||||
investigation_id TEXT NOT NULL,
|
||||
source_entity_id TEXT NOT NULL,
|
||||
target_entity_id TEXT NOT NULL,
|
||||
relationship_type TEXT NOT NULL,
|
||||
confidence REAL DEFAULT 1.0,
|
||||
evidence TEXT DEFAULT '{}',
|
||||
FOREIGN KEY (investigation_id) REFERENCES investigations(id),
|
||||
FOREIGN KEY (source_entity_id) REFERENCES entities(id),
|
||||
FOREIGN KEY (target_entity_id) REFERENCES entities(id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_entities_investigation ON entities(investigation_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_relationships_investigation ON relationships(investigation_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_entities_type ON entities(entity_type);
|
||||
CREATE INDEX IF NOT EXISTS idx_entities_value ON entities(value);
|
||||
"""
|
||||
|
||||
|
||||
class Database:
|
||||
"""SQLite database manager for EireScope investigations."""
|
||||
|
||||
def __init__(self, db_path: str):
|
||||
self.db_path = db_path
|
||||
os.makedirs(os.path.dirname(db_path), exist_ok=True)
|
||||
self._init_db()
|
||||
|
||||
def _init_db(self):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
conn.executescript(CREATE_TABLES_SQL)
|
||||
conn.commit()
|
||||
logger.info(f"Database initialized at {self.db_path}")
|
||||
|
||||
def _conn(self) -> sqlite3.Connection:
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
def save_investigation(self, inv: Investigation):
|
||||
"""Save or update an investigation and all its entities/relationships."""
|
||||
with self._conn() as conn:
|
||||
conn.execute(
|
||||
"""INSERT OR REPLACE INTO investigations
|
||||
(id, initial_query, initial_type, status, notes, modules_run, created_at, completed_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
||||
(inv.id, inv.initial_query, inv.initial_type.value, inv.status,
|
||||
inv.notes, json.dumps(inv.modules_run), inv.created_at, inv.completed_at),
|
||||
)
|
||||
for entity in inv.entities:
|
||||
conn.execute(
|
||||
"""INSERT OR REPLACE INTO entities
|
||||
(id, investigation_id, entity_type, value, source_module, confidence, metadata, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
||||
(entity.id, inv.id, entity.entity_type.value, entity.value,
|
||||
entity.source_module, entity.confidence,
|
||||
json.dumps(entity.metadata), entity.created_at),
|
||||
)
|
||||
for rel in inv.relationships:
|
||||
conn.execute(
|
||||
"""INSERT OR REPLACE INTO relationships
|
||||
(id, investigation_id, source_entity_id, target_entity_id,
|
||||
relationship_type, confidence, evidence)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
||||
(rel.id, inv.id, rel.source_entity_id, rel.target_entity_id,
|
||||
rel.relationship_type, rel.confidence, json.dumps(rel.evidence)),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
def load_investigation(self, inv_id: str) -> Optional[Investigation]:
|
||||
"""Load a full investigation by ID."""
|
||||
with self._conn() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT * FROM investigations WHERE id = ?", (inv_id,)
|
||||
).fetchone()
|
||||
if not row:
|
||||
return None
|
||||
|
||||
inv = Investigation(
|
||||
id=row["id"],
|
||||
initial_query=row["initial_query"],
|
||||
initial_type=EntityType(row["initial_type"]),
|
||||
status=row["status"],
|
||||
notes=row["notes"] or "",
|
||||
modules_run=json.loads(row["modules_run"]),
|
||||
created_at=row["created_at"],
|
||||
completed_at=row["completed_at"],
|
||||
)
|
||||
|
||||
for erow in conn.execute(
|
||||
"SELECT * FROM entities WHERE investigation_id = ?", (inv_id,)
|
||||
):
|
||||
inv.entities.append(Entity(
|
||||
id=erow["id"],
|
||||
entity_type=EntityType(erow["entity_type"]),
|
||||
value=erow["value"],
|
||||
source_module=erow["source_module"],
|
||||
confidence=erow["confidence"],
|
||||
metadata=json.loads(erow["metadata"]),
|
||||
created_at=erow["created_at"],
|
||||
))
|
||||
|
||||
for rrow in conn.execute(
|
||||
"SELECT * FROM relationships WHERE investigation_id = ?", (inv_id,)
|
||||
):
|
||||
inv.relationships.append(EntityRelationship(
|
||||
id=rrow["id"],
|
||||
source_entity_id=rrow["source_entity_id"],
|
||||
target_entity_id=rrow["target_entity_id"],
|
||||
relationship_type=rrow["relationship_type"],
|
||||
confidence=rrow["confidence"],
|
||||
evidence=json.loads(rrow["evidence"]),
|
||||
))
|
||||
|
||||
return inv
|
||||
|
||||
def list_investigations(self, limit: int = 50) -> List[Dict]:
|
||||
"""List recent investigations (summary only)."""
|
||||
with self._conn() as conn:
|
||||
rows = conn.execute(
|
||||
"""SELECT i.*, COUNT(e.id) as entity_count
|
||||
FROM investigations i
|
||||
LEFT JOIN entities e ON e.investigation_id = i.id
|
||||
GROUP BY i.id
|
||||
ORDER BY i.created_at DESC LIMIT ?""",
|
||||
(limit,),
|
||||
).fetchall()
|
||||
return [
|
||||
{
|
||||
"id": r["id"],
|
||||
"initial_query": r["initial_query"],
|
||||
"initial_type": r["initial_type"],
|
||||
"status": r["status"],
|
||||
"entity_count": r["entity_count"],
|
||||
"created_at": r["created_at"],
|
||||
"completed_at": r["completed_at"],
|
||||
}
|
||||
for r in rows
|
||||
]
|
||||
|
||||
def delete_investigation(self, inv_id: str):
|
||||
"""Delete an investigation and all associated data."""
|
||||
with self._conn() as conn:
|
||||
conn.execute("DELETE FROM relationships WHERE investigation_id = ?", (inv_id,))
|
||||
conn.execute("DELETE FROM entities WHERE investigation_id = ?", (inv_id,))
|
||||
conn.execute("DELETE FROM investigations WHERE id = ?", (inv_id,))
|
||||
conn.commit()
|
||||
0
eirescope/modules/__init__.py
Normal file
0
eirescope/modules/__init__.py
Normal file
52
eirescope/modules/base.py
Normal file
52
eirescope/modules/base.py
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
"""Base class for all EireScope OSINT modules."""
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import List, Dict, Any
|
||||
from eirescope.core.entity import Entity, EntityType, Investigation
|
||||
|
||||
|
||||
class BaseOSINTModule(ABC):
|
||||
"""Abstract base for all OSINT plugins.
|
||||
|
||||
Every module must define:
|
||||
- name: human-readable module name
|
||||
- description: what this module does
|
||||
- supported_entity_types: which entity types it can process
|
||||
- requires_api_key: whether an API key is needed
|
||||
"""
|
||||
|
||||
name: str = "Base Module"
|
||||
description: str = ""
|
||||
supported_entity_types: List[EntityType] = []
|
||||
requires_api_key: bool = False
|
||||
api_key_name: str = ""
|
||||
icon: str = "search"
|
||||
|
||||
def __init__(self, config: Dict[str, Any] = None):
|
||||
self.config = config or {}
|
||||
self.api_key = self.config.get(self.api_key_name, "")
|
||||
|
||||
def can_handle(self, entity_type: EntityType) -> bool:
|
||||
"""Check if this module supports the given entity type."""
|
||||
return entity_type in self.supported_entity_types
|
||||
|
||||
@abstractmethod
|
||||
def execute(self, entity: Entity, investigation: Investigation) -> List[Entity]:
|
||||
"""Execute the OSINT module against an entity.
|
||||
|
||||
Args:
|
||||
entity: The entity to investigate.
|
||||
investigation: The parent investigation (for adding relationships).
|
||||
|
||||
Returns:
|
||||
List of newly discovered entities.
|
||||
"""
|
||||
pass
|
||||
|
||||
def get_metadata(self) -> Dict:
|
||||
return {
|
||||
"name": self.name,
|
||||
"description": self.description,
|
||||
"supported_types": [t.value for t in self.supported_entity_types],
|
||||
"requires_api_key": self.requires_api_key,
|
||||
"icon": self.icon,
|
||||
}
|
||||
0
eirescope/utils/__init__.py
Normal file
0
eirescope/utils/__init__.py
Normal file
31
eirescope/utils/exceptions.py
Normal file
31
eirescope/utils/exceptions.py
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
"""Custom exceptions for EireScope."""
|
||||
|
||||
|
||||
class EireScopeError(Exception):
|
||||
"""Base exception for EireScope."""
|
||||
pass
|
||||
|
||||
|
||||
class ValidationError(EireScopeError):
|
||||
"""Invalid input data."""
|
||||
pass
|
||||
|
||||
|
||||
class ModuleError(EireScopeError):
|
||||
"""Error in an OSINT module."""
|
||||
pass
|
||||
|
||||
|
||||
class ModuleNotFoundError(EireScopeError):
|
||||
"""Requested module not found."""
|
||||
pass
|
||||
|
||||
|
||||
class RateLimitError(EireScopeError):
|
||||
"""Rate limited by external service."""
|
||||
pass
|
||||
|
||||
|
||||
class APIKeyRequiredError(EireScopeError):
|
||||
"""Module requires an API key that is not configured."""
|
||||
pass
|
||||
105
eirescope/utils/http_client.py
Normal file
105
eirescope/utils/http_client.py
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
"""HTTP client with retries, rate limiting, and user-agent rotation for OSINT."""
|
||||
import time
|
||||
import random
|
||||
import logging
|
||||
import requests
|
||||
from typing import Optional, Dict, Any
|
||||
|
||||
logger = logging.getLogger("eirescope.http")
|
||||
|
||||
USER_AGENTS = [
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||||
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:121.0) Gecko/20100101 Firefox/121.0",
|
||||
]
|
||||
|
||||
|
||||
class OSINTHTTPClient:
|
||||
"""HTTP client optimized for OSINT data collection."""
|
||||
|
||||
def __init__(self, timeout: int = 10, max_retries: int = 3,
|
||||
rate_limit: float = 0.5, proxy: str = None):
|
||||
self.timeout = timeout
|
||||
self.max_retries = max_retries
|
||||
self.rate_limit = rate_limit
|
||||
self.session = requests.Session()
|
||||
if proxy:
|
||||
self.session.proxies = {"http": proxy, "https": proxy}
|
||||
self._last_request_time = 0
|
||||
|
||||
def _get_headers(self, extra_headers: Dict = None) -> Dict:
|
||||
headers = {
|
||||
"User-Agent": random.choice(USER_AGENTS),
|
||||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||||
"Accept-Language": "en-US,en;q=0.5",
|
||||
"Accept-Encoding": "gzip, deflate",
|
||||
"DNT": "1",
|
||||
"Connection": "keep-alive",
|
||||
}
|
||||
if extra_headers:
|
||||
headers.update(extra_headers)
|
||||
return headers
|
||||
|
||||
def _rate_limit_wait(self):
|
||||
elapsed = time.time() - self._last_request_time
|
||||
if elapsed < self.rate_limit:
|
||||
time.sleep(self.rate_limit - elapsed)
|
||||
self._last_request_time = time.time()
|
||||
|
||||
def get(self, url: str, headers: Dict = None, params: Dict = None,
|
||||
allow_redirects: bool = True, **kwargs) -> Optional[requests.Response]:
|
||||
return self._request("GET", url, headers=headers, params=params,
|
||||
allow_redirects=allow_redirects, **kwargs)
|
||||
|
||||
def head(self, url: str, headers: Dict = None,
|
||||
allow_redirects: bool = True, **kwargs) -> Optional[requests.Response]:
|
||||
return self._request("HEAD", url, headers=headers,
|
||||
allow_redirects=allow_redirects, **kwargs)
|
||||
|
||||
def post(self, url: str, headers: Dict = None, data: Any = None,
|
||||
json: Any = None, **kwargs) -> Optional[requests.Response]:
|
||||
return self._request("POST", url, headers=headers, data=data,
|
||||
json=json, **kwargs)
|
||||
|
||||
def _request(self, method: str, url: str, **kwargs) -> Optional[requests.Response]:
|
||||
extra_headers = kwargs.pop("headers", None)
|
||||
kwargs["headers"] = self._get_headers(extra_headers)
|
||||
kwargs.setdefault("timeout", self.timeout)
|
||||
|
||||
for attempt in range(self.max_retries):
|
||||
try:
|
||||
self._rate_limit_wait()
|
||||
response = self.session.request(method, url, **kwargs)
|
||||
return response
|
||||
except requests.exceptions.Timeout:
|
||||
logger.warning(f"Timeout on {method} {url} (attempt {attempt + 1})")
|
||||
except requests.exceptions.ConnectionError:
|
||||
logger.warning(f"Connection error on {method} {url} (attempt {attempt + 1})")
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.error(f"Request error on {method} {url}: {e}")
|
||||
return None
|
||||
|
||||
if attempt < self.max_retries - 1:
|
||||
backoff = (2 ** attempt) + random.uniform(0, 1)
|
||||
time.sleep(backoff)
|
||||
|
||||
logger.error(f"All {self.max_retries} retries failed for {method} {url}")
|
||||
return None
|
||||
|
||||
def check_url_exists(self, url: str, timeout: int = 5) -> bool:
|
||||
"""Quick check if a URL returns a successful response."""
|
||||
try:
|
||||
self._rate_limit_wait()
|
||||
resp = self.session.get(
|
||||
url,
|
||||
headers=self._get_headers(),
|
||||
timeout=timeout,
|
||||
allow_redirects=True,
|
||||
stream=True,
|
||||
)
|
||||
resp.close()
|
||||
return resp.status_code == 200
|
||||
except Exception:
|
||||
return False
|
||||
98
eirescope/utils/validators.py
Normal file
98
eirescope/utils/validators.py
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
"""Input validation and normalization for EireScope."""
|
||||
import re
|
||||
import socket
|
||||
from typing import Tuple, Optional
|
||||
from eirescope.core.entity import EntityType
|
||||
|
||||
|
||||
class EntityValidator:
|
||||
"""Validates and normalizes user-provided search inputs."""
|
||||
|
||||
EMAIL_RE = re.compile(
|
||||
r"^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$"
|
||||
)
|
||||
USERNAME_RE = re.compile(r"^[a-zA-Z0-9._\-]{1,64}$")
|
||||
PHONE_RE = re.compile(r"^\+?[\d\s\-\(\)]{7,20}$")
|
||||
DOMAIN_RE = re.compile(
|
||||
r"^(?:[a-zA-Z0-9](?:[a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)+"
|
||||
r"[a-zA-Z]{2,}$"
|
||||
)
|
||||
IP_V4_RE = re.compile(
|
||||
r"^(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}"
|
||||
r"(?:25[0-5]|2[0-4]\d|[01]?\d\d?)$"
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def validate_email(cls, value: str) -> bool:
|
||||
return bool(cls.EMAIL_RE.match(value.strip()))
|
||||
|
||||
@classmethod
|
||||
def validate_username(cls, value: str) -> bool:
|
||||
return bool(cls.USERNAME_RE.match(value.strip()))
|
||||
|
||||
@classmethod
|
||||
def validate_phone(cls, value: str) -> bool:
|
||||
cleaned = re.sub(r"[\s\-\(\)]", "", value.strip())
|
||||
return bool(re.match(r"^\+?\d{7,15}$", cleaned))
|
||||
|
||||
@classmethod
|
||||
def validate_domain(cls, value: str) -> bool:
|
||||
return bool(cls.DOMAIN_RE.match(value.strip().lower()))
|
||||
|
||||
@classmethod
|
||||
def validate_ip(cls, value: str) -> bool:
|
||||
v = value.strip()
|
||||
if cls.IP_V4_RE.match(v):
|
||||
return True
|
||||
try:
|
||||
socket.inet_pton(socket.AF_INET6, v)
|
||||
return True
|
||||
except (socket.error, OSError):
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def normalize(cls, value: str, entity_type: EntityType) -> str:
|
||||
"""Normalize input based on entity type."""
|
||||
v = value.strip()
|
||||
if entity_type == EntityType.EMAIL:
|
||||
return v.lower()
|
||||
if entity_type == EntityType.DOMAIN:
|
||||
return v.lower().rstrip(".")
|
||||
if entity_type == EntityType.PHONE:
|
||||
return re.sub(r"[\s\-\(\)]", "", v)
|
||||
if entity_type == EntityType.IP_ADDRESS:
|
||||
return v
|
||||
if entity_type == EntityType.USERNAME:
|
||||
return v.lstrip("@")
|
||||
return v
|
||||
|
||||
@classmethod
|
||||
def detect_type(cls, value: str) -> Optional[EntityType]:
|
||||
"""Auto-detect entity type from input value."""
|
||||
v = value.strip()
|
||||
if cls.validate_email(v):
|
||||
return EntityType.EMAIL
|
||||
if cls.validate_ip(v):
|
||||
return EntityType.IP_ADDRESS
|
||||
if cls.validate_phone(v) and (v.startswith("+") or len(re.sub(r"\D", "", v)) >= 10):
|
||||
return EntityType.PHONE
|
||||
if cls.validate_domain(v):
|
||||
return EntityType.DOMAIN
|
||||
if cls.validate_username(v):
|
||||
return EntityType.USERNAME
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def validate_and_normalize(cls, value: str, entity_type: EntityType) -> Tuple[bool, str]:
|
||||
"""Validate and normalize input. Returns (is_valid, normalized_value)."""
|
||||
validators = {
|
||||
EntityType.EMAIL: cls.validate_email,
|
||||
EntityType.USERNAME: cls.validate_username,
|
||||
EntityType.PHONE: cls.validate_phone,
|
||||
EntityType.DOMAIN: cls.validate_domain,
|
||||
EntityType.IP_ADDRESS: cls.validate_ip,
|
||||
}
|
||||
validator = validators.get(entity_type)
|
||||
if validator and not validator(value):
|
||||
return False, value
|
||||
return True, cls.normalize(value, entity_type)
|
||||
Loading…
Add table
Add a link
Reference in a new issue