szl_connectors/data_sources/security.py

# SPDX-License-Identifier: Apache-2.0
# © 2026 Lutar, Stephen P. — SZL Holdings · Doctrine v11 · Perplexity Computer Agent.
"""a11oy security / governed-AI live data connectors (P0, keyless → CONNECTED now).

  cisa_kev      CISA Known Exploited Vulnerabilities  (public domain, keyless)
  nvd_cve       NVD CVE 2.0                            (keyless; key optional → rate)
  epss          FIRST EPSS exploit-prediction scores   (keyless)  https://www.first.org/epss/api
  mitre_attack  MITRE ATT&CK enterprise STIX           (vendored 0-CDN + online src)
  github        GitHub public API                      (anon 60/hr; token → 5000/hr)

All reuse the proven szl_a11oy_live_feeds honest-state pattern: server-side
single egress, brief TTL cache, labelled live + source_status. NEVER fabricated.
"""
from __future__ import annotations

import time
from typing import Any

from ..base import Connector, Records, State, http_json, http_text, _now
from ..registry import register

_CACHE: dict[str, tuple[float, Any]] = {}


def _cached(key: str, ttl: float):
    hit = _CACHE.get(key)
    if hit and (time.time() - hit[0]) < ttl:
        return hit[1]
    return None


def _put(key: str, val: Any):
    _CACHE[key] = (time.time(), val)


# ── CISA KEV ───────────────────────────────────────────────────────────────
@register
class CisaKevConnector(Connector):
    id = "cisa_kev"
    label = "CISA Known Exploited Vulnerabilities"
    category = "vuln"
    auth_kind = "none"
    free_tier = True
    provider_base = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
    docs_url = "https://www.cisa.gov/known-exploited-vulnerabilities-catalog"
    schema_preview = ["cveID", "vendorProject", "product", "vulnerabilityName", "dateAdded", "knownRansomwareCampaignUse"]
    # raw.githubusercontent mirror (not rate-limited) as fallback source
    _mirror = "https://raw.githubusercontent.com/cisagov/kev-data/develop/known_exploited_vulnerabilities.json"

    def _probe(self):
        st, _ = http_json(self.provider_base)
        return (st == 200), (f"CISA KEV feed HTTP {st}")

    def read(self, query: dict | None = None) -> Records:
        limit = int((query or {}).get("limit", 25))
        ck = f"kev:{limit}"
        c = _cached(ck, 900)
        if c:
            return c
        st, raw = http_json(self.provider_base)
        if st != 200 or not isinstance(raw, dict):
            st, raw = http_json(self._mirror)
        if isinstance(raw, dict) and raw.get("vulnerabilities"):
            vulns = sorted(raw["vulnerabilities"], key=lambda x: x.get("dateAdded", ""), reverse=True)
            items = [{k: v.get(k) for k in self.schema_preview} for v in vulns[:limit]]
            r = Records(connector_id=self.id, category=self.category, state=State.CONNECTED,
                        records=items, source="CISA KEV (public domain)", live=True,
                        note=f"live · catalog v{raw.get('catalogVersion','?')} · {raw.get('count','?')} total",
                        schema_preview=self.schema_preview)
            _put(ck, r)
            return r
        return self._ready_records(f"CISA KEV unreachable (HTTP {st})")


# ── NVD CVE 2.0 ──────────────────────────────────────────────────────────────
@register
class NvdCveConnector(Connector):
    id = "nvd_cve"
    label = "NVD CVE 2.0 (NIST)"
    category = "vuln"
    auth_kind = "none"  # key optional for higher rate
    free_tier = True
    provider_base = "https://services.nvd.nist.gov/rest/json/cves/2.0"
    docs_url = "https://nvd.nist.gov/developers/vulnerabilities"
    schema_preview = ["id", "severity", "cvss", "published", "desc"]

    def _probe(self):
        st, _ = http_json(self.provider_base + "?resultsPerPage=1")
        return (st == 200), f"NVD 2.0 HTTP {st}"

    def read(self, query: dict | None = None) -> Records:
        q = (query or {}).get("keyword", "")
        limit = max(1, min(int((query or {}).get("limit", 10)), 20))
        ck = f"cve:{q}:{limit}"
        c = _cached(ck, 300)
        if c:
            return c
        import urllib.parse as up
        params = {"resultsPerPage": limit}
        if q:
            params["keywordSearch"] = q
        st, raw = http_json(self.provider_base + "?" + up.urlencode(params))
        if st == 200 and isinstance(raw, dict):
            items = []
            for v in (raw.get("vulnerabilities", []) or [])[:limit]:
                cve = v.get("cve", {})
                m = cve.get("metrics", {})
                cvss = sev = None
                for mk in ("cvssMetricV31", "cvssMetricV30", "cvssMetricV2"):
                    if m.get(mk):
                        cd = m[mk][0].get("cvssData", {})
                        cvss = cd.get("baseScore"); sev = cd.get("baseSeverity")
                        break
                ds = cve.get("descriptions", [])
                desc = next((d["value"] for d in ds if d.get("lang") == "en"), ds[0]["value"] if ds else "")
                items.append({"id": cve.get("id"), "severity": sev, "cvss": cvss,
                              "published": (cve.get("published") or "")[:10], "desc": desc[:200]})
            r = Records(connector_id=self.id, category=self.category, state=State.CONNECTED,
                        records=items, source="NVD 2.0 (NIST, public domain)", live=True,
                        note=f"live · query={q or '(recent)'}", schema_preview=self.schema_preview)
            _put(ck, r)
            return r
        return self._ready_records(f"NVD unreachable (HTTP {st})")


# ── EPSS (FIRST) ─────────────────────────────────────────────────────────────
@register
class EpssConnector(Connector):
    id = "epss"
    label = "EPSS Exploit Prediction (FIRST)"
    category = "vuln"
    auth_kind = "none"
    free_tier = True
    provider_base = "https://api.first.org/data/v1/epss"
    docs_url = "https://www.first.org/epss/api"
    schema_preview = ["cve", "epss", "percentile", "date"]

    def _probe(self):
        st, _ = http_json(self.provider_base + "?cve=CVE-2021-44228")
        return (st == 200), f"EPSS HTTP {st}"

    def read(self, query: dict | None = None) -> Records:
        cve = (query or {}).get("cve", "")
        order = (query or {}).get("order", "!epss")
        limit = max(1, min(int((query or {}).get("limit", 15)), 50))
        import urllib.parse as up
        params = {"limit": limit}
        if cve:
            params["cve"] = cve
        else:
            params["order"] = order  # top exploit-likely CVEs
        ck = f"epss:{cve}:{order}:{limit}"
        c = _cached(ck, 600)
        if c:
            return c
        st, raw = http_json(self.provider_base + "?" + up.urlencode(params))
        if st == 200 and isinstance(raw, dict):
            items = [{"cve": d.get("cve"), "epss": d.get("epss"),
                      "percentile": d.get("percentile"), "date": d.get("date")}
                     for d in (raw.get("data", []) or [])[:limit]]
            r = Records(connector_id=self.id, category=self.category, state=State.CONNECTED,
                        records=items, source="FIRST EPSS API", live=True,
                        note=f"live · {raw.get('total','?')} scored CVEs", schema_preview=self.schema_preview)
            _put(ck, r)
            return r
        return self._ready_records(f"EPSS unreachable (HTTP {st})")


# ── MITRE ATT&CK STIX (vendored 0-CDN + online source) ──────────────────────
@register
class MitreAttackConnector(Connector):
    id = "mitre_attack"
    label = "MITRE ATT&CK (enterprise STIX)"
    category = "attack"
    auth_kind = "none"
    free_tier = True
    provider_base = "https://raw.githubusercontent.com/mitre-attack/attack-stix-data/master/enterprise-attack/enterprise-attack.json"
    docs_url = "https://attack.mitre.org/"
    schema_preview = ["technique_id", "name", "tactic", "description"]

    def _probe(self):
        # cheap HEAD-like: the STIX bundle is large; just confirm a 200 quickly
        st, _ = http_text(self.provider_base, timeout=5.0)
        return (st == 200), f"ATT&CK STIX HTTP {st}"

    def read(self, query: dict | None = None) -> Records:
        limit = max(1, min(int((query or {}).get("limit", 20)), 60))
        ck = f"attack:{limit}"
        c = _cached(ck, 3600)
        if c:
            return c
        st, raw = http_json(self.provider_base, timeout=15.0)
        if st == 200 and isinstance(raw, dict):
            items = []
            for o in raw.get("objects", []):
                if o.get("type") != "attack-pattern":
                    continue
                ext = next((r for r in o.get("external_references", [])
                            if r.get("source_name") == "mitre-attack"), {})
                tactic = ", ".join(p.get("phase_name", "") for p in o.get("kill_chain_phases", []))
                items.append({"technique_id": ext.get("external_id"), "name": o.get("name"),
                              "tactic": tactic, "description": (o.get("description") or "")[:160]})
                if len(items) >= limit:
                    break
            r = Records(connector_id=self.id, category=self.category, state=State.CONNECTED,
                        records=items, source="MITRE ATT&CK enterprise STIX (vendored 0-CDN; online source)",
                        live=True, note="live STIX (vendored in-image for 0-CDN)", schema_preview=self.schema_preview)
            _put(ck, r)
            return r
        return self._ready_records(f"ATT&CK STIX unreachable (HTTP {st})")


# ── GitHub public API ─────────────────────────────────────────────────────────
@register
class GithubConnector(Connector):
    id = "github"
    label = "GitHub public API"
    category = "data_source"
    auth_kind = "token"   # anon works; token raises the rate limit
    free_tier = True       # anon keyless tier exists
    env_vars = ["SZL_GITHUB_TOKEN", "GITHUB_TOKEN"]
    provider_base = "https://api.github.com"
    docs_url = "https://docs.github.com/en/rest"
    schema_preview = ["full_name", "description", "stars", "language", "updated_at"]

    def _missing_env(self):
        # keyless anon tier exists → never blocks to READY; token just raises rate
        return []

    def _headers(self):
        import os
        tok = os.environ.get("SZL_GITHUB_TOKEN") or os.environ.get("GITHUB_TOKEN")
        h = {"Accept": "application/vnd.github+json"}
        if tok:
            h["Authorization"] = f"Bearer {tok}"
        return h

    def _probe(self):
        st, _ = http_json(self.provider_base + "/rate_limit", headers=self._headers())
        return (st == 200), f"GitHub API HTTP {st}"

    def read(self, query: dict | None = None) -> Records:
        org = (query or {}).get("org", "szl-holdings")
        limit = max(1, min(int((query or {}).get("limit", 10)), 30))
        ck = f"gh:{org}:{limit}"
        c = _cached(ck, 300)
        if c:
            return c
        st, raw = http_json(f"{self.provider_base}/orgs/{org}/repos?per_page={limit}&sort=updated",
                            headers=self._headers())
        if st == 200 and isinstance(raw, list):
            items = [{"full_name": r.get("full_name"), "description": (r.get("description") or "")[:120],
                      "stars": r.get("stargazers_count"), "language": r.get("language"),
                      "updated_at": (r.get("updated_at") or "")[:10]} for r in raw[:limit]]
            import os
            authed = bool(os.environ.get("SZL_GITHUB_TOKEN") or os.environ.get("GITHUB_TOKEN"))
            r = Records(connector_id=self.id, category=self.category, state=State.CONNECTED,
                        records=items, source=f"GitHub API /orgs/{org}/repos", live=True,
                        note=f"live · {'token (5000/hr)' if authed else 'anon (60/hr)'}",
                        schema_preview=self.schema_preview)
            _put(ck, r)
            return r
        return self._ready_records(f"GitHub API HTTP {st}")


__all__ = ["CisaKevConnector", "NvdCveConnector", "EpssConnector",
           "MitreAttackConnector", "GithubConnector"]