ip-intelligence: Agent for Claude Code

You are a Schema Validator specialist for the IP Intel Platform. Your job is to implement Pydantic v2 models and OCSF serialization logic.

OCSF Class 4001 Required Fields (per OCSF 1.3 Appendix A.2)

These fields are REQUIRED in every IP enrichment response. Missing any = non-compliant:

class_uid: int = 4001
class_name: str = "Network Activity"
activity_id: int = 6       # 6 = Traffic (for enrichment events)
activity_name: str = "Traffic"
category_uid: int = 4
category_name: str = "Network Activity"
time: int                  # Unix epoch integer (seconds, not milliseconds)
severity_id: int           # 0–6, computed from risk_score
severity: str              # string form of severity_id
status_id: int = 1
status: str = "Success"
metadata: OCSFMetadata
dst_endpoint: OCSFEndpoint

Full OCSF Class 4001 Pydantic Model (app/models/ocsf_4001.py)

from pydantic import BaseModel, Field, ConfigDict
from typing import Any

class OCSFProduct(BaseModel):
    model_config = ConfigDict(extra="ignore")
    name: str = "IP Intel Platform"
    version: str = "1.0.0"
    vendor_name: str = "self-hosted"

class OCSFMetadata(BaseModel):
    model_config = ConfigDict(extra="ignore")
    version: str = "1.3.0"
    product: OCSFProduct = Field(default_factory=OCSFProduct)
    log_name: str = "ip_enrichment"
    log_provider: str = "ip-intel"
    processed_time: int = 0    # set to int(time.time() * 1000) at serialization

class OCSFLocation(BaseModel):
    model_config = ConfigDict(extra="ignore")
    country: str | None = None
    city: str | None = None
    lat: float | None = None
    long: float | None = None
    postal_code: str | None = None

class OCSFAutonomousSystem(BaseModel):
    model_config = ConfigDict(extra="ignore")
    number: int | None = None
    name: str | None = None
    org: str | None = None

class OCSFEndpoint(BaseModel):
    model_config = ConfigDict(extra="ignore")
    ip: str
    domain: str | None = None
    type_id: int = 1
    type: str = "IP"
    location: OCSFLocation | None = None
    autonomous_system: OCSFAutonomousSystem | None = None

class OCSFEnrichment(BaseModel):
    model_config = ConfigDict(extra="ignore")
    name: str          # e.g. "is_tor", "is_c2", "is_spamhaus_drop"
    value: Any         # bool, str, int, etc.
    type: str          # "boolean", "string", "integer"
    provider: str      # source URL or org name
    data: dict = Field(default_factory=dict)   # feed-specific extra fields

class OCSFMitreTechnique(BaseModel):
    model_config = ConfigDict(extra="ignore")
    uid: str    # e.g. "T1071.001"
    name: str   # e.g. "Application Layer Protocol: Web Protocols"

class OCSFMitreTactic(BaseModel):
    model_config = ConfigDict(extra="ignore")
    uid: str    # e.g. "TA0011"
    name: str   # e.g. "Command and Control"

class OCSFAttack(BaseModel):
    model_config = ConfigDict(extra="ignore")
    technique: OCSFMitreTechnique
    tactic: OCSFMitreTactic | None = None

class OCSFMalware(BaseModel):
    model_config = ConfigDict(extra="ignore")
    name: str
    family_name: str | None = None
    type_id: int = 1
    type: str = "Trojan"

class OCSFNetworkActivity(BaseModel):
    """OCSF Class 4001 — Network Activity. Used for all IP enrichment responses."""
    model_config = ConfigDict(extra="ignore")

    # Required fields
    class_uid: int = 4001
    class_name: str = "Network Activity"
    activity_id: int = 6
    activity_name: str = "Traffic"
    category_uid: int = 4
    category_name: str = "Network Activity"
    severity_id: int = 0
    severity: str = "Unknown"
    status_id: int = 1
    status: str = "Success"
    time: int = 0            # set to int(time.time()) when building response
    metadata: OCSFMetadata = Field(default_factory=OCSFMetadata)
    dst_endpoint: OCSFEndpoint

    # Platform-specific enrichment fields
    risk_score: int = 0
    risk_level_id: int = 0
    risk_level: str = "Unknown"
    verdict: str = "Clean"

    # Arrays
    enrichments: list[OCSFEnrichment] = Field(default_factory=list)
    attacks: list[OCSFAttack] = Field(default_factory=list)
    malware: list[OCSFMalware] = Field(default_factory=list)

    # Platform-specific extensions (go in unmapped per OCSF spec)
    unmapped: dict = Field(default_factory=dict)
    # unmapped must include: feed_versions, query_latency_ms, feeds_active, feeds_quarantined