github · mcp-server16c78a447a0a

Evidence

WorkingDirectory=/app

EnvironmentFile=/app/.env

ExecStart=/app/venv/bin/python3 -m uvicorn datanexus.main:app \

    --host 0.0.0.0 --port 8000 --workers 2

Restart=always

RestartSec=5

StandardOutput=journal

Remediation

Bind to 127.0.0.1 for local-only access. If cross-host access is truly required, put the service behind an authenticated reverse proxy rather than exposing it on 0.0.0.0.

Evidence

# Start feedback dashboard in background (port 8101)

uvicorn feedback.dashboard.server:app \

  --host 0.0.0.0 \

  --port 8101 \

  --log-level info &

DASHBOARD_PID=$!

Remediation

Bind to 127.0.0.1 for local-only access. If cross-host access is truly required, put the service behind an authenticated reverse proxy rather than exposing it on 0.0.0.0.

Evidence

if __name__ == "__main__":

    import uvicorn

    logging.basicConfig(level=logging.INFO)

    uvicorn.run("feedback.dashboard.server:app", host="0.0.0.0", port=_PORT, reload=False)

Remediation

Bind to 127.0.0.1 for local-only access. If cross-host access is truly required, put the service behind an authenticated reverse proxy rather than exposing it on 0.0.0.0.

Evidence

WorkingDirectory=/app

EnvironmentFile=/app/.env

ExecStart=/app/venv/bin/python3 -m uvicorn feedback.dashboard.server:app \

    --host 0.0.0.0 --port 8101 --workers 1

Restart=always

RestartSec=5

StandardOutput=journal

Remediation

Bind to 127.0.0.1 for local-only access. If cross-host access is truly required, put the service behind an authenticated reverse proxy rather than exposing it on 0.0.0.0.

Evidence

)

    main.run(

        transport="streamable-http",

        host="0.0.0.0",   # nosec B104

        port=8000,

        middleware=[Middleware(_ClientIPMiddleware), Middleware(_ApiKeyMiddleware)],

        stateless_http=True,

Remediation

Bind to 127.0.0.1 for local-only access. If cross-host access is truly required, put the service behind an authenticated reverse proxy rather than exposing it on 0.0.0.0.

Evidence

# ── Structured error response shape ──────────────────────────────────────────

# Return this dict on ANY error. Never raise. Never return str(e).

# error_code must come from ErrorCode enum below.

#

# {

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

"error":           str(exc),

            "haiku_available": False,

        }))

        return {"error": str(exc), "haiku_available": False}

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

schedule_seconds=_SCHEDULE,

        )

    async def fetch(self) -> bytes:

        """Fetch and cache EPO OPS OAuth token; probe fallback sources."""

        client_id     = os.environ.get("EPO_CLIENT_ID", "")

        client_secret = os.environ.get("EPO_CLIENT_SECRET", "")

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

schedule_seconds=_SCHEDULE,

        )

    async def fetch(self) -> bytes:

        """

        Fetch NPPES data for top specialities. Returns a summary JSON blob

        for the base class to hash and store. Per-speciality cache entries

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

schedule_seconds=_SCHEDULE,

        )

    async def fetch(self) -> bytes:

        """Pre-seed top keyword categories."""

        seeded = 0

        async with httpx.AsyncClient(

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

self.ttl_seconds      = ttl_seconds

        self.schedule_seconds = schedule_seconds

    async def fetch(self) -> bytes:

        """

        Override in subclass. Must return raw upstream response bytes.

        Raise any exception on failure — run_forever will catch it.

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

schedule_seconds=604800,  # weekly — data updates monthly

        )

    async def fetch(self) -> bytes:

        """Download all 4 regional CSVs and index every EIN into Redis."""

        total_rows = 0

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

schedule_seconds=_SCHEDULE,  # 21600 — rate limit awareness

        )

    async def fetch(self) -> bytes:

        """Pre-seed top regulatory keyword categories."""

        api_key = os.environ.get("REGULATIONS_GOV_KEY", "")

        if not api_key:

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

Downloads from the official CISA GitHub mirror (cisagov/kev-data) — see KEV_URL constant.

and stores in Redis as:

  datanexus:kev:catalog    — full catalog JSON (TTL 25h)

  datanexus:kev:fetched_at — ISO timestamp of last successful fetch (TTL 25h)

  datanexus:kev:last_refresh_error — error message if refresh fails (TTL 7d)

Invoked two ways:

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

Endpoints:

  GET /ops/dashboard          — full HTML dashboard

  GET /ops/api/metrics        — JSON metrics snapshot (same data, for fetch())

  GET /ops/health             — {"ok": true}

The dashboard auto-refreshes every 15 seconds via JavaScript fetch().

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

class IngestBase:

    """

    Base ingest worker. Subclasses override fetch() only.

    Args:

        tool_id:          Tool this worker feeds (e.g. 'T04')

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

schedule_seconds=_SCHEDULE,

        )

    async def fetch(self) -> bytes:

        """Fetch IANA RDAP bootstrap and store in Redis."""

        async with httpx.AsyncClient(

            timeout=_TIMEOUT, headers=_HEADERS, follow_redirects=True

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

schedule_seconds=3600,

        )

    async def fetch(self) -> bytes:

        """Pre-seed all 50 packages. Returns last raw response bytes."""

        seeded = 0

        last_raw = b""

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

schedule_seconds=86400,

        )

    async def fetch(self) -> bytes:

        """Download bulk extract and index all main charities into Redis."""

        log.info(json.dumps({

            "ts": _iso_now(), "event": "uk_bulk_download_start", "tool": self.tool_id,

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

except Exception as _cb_exc:

                    log.warning("ingest_base.circuit_check error: %s", _cb_exc)

                raw: bytes = await self.fetch()

                payload_hash = compute_payload_hash(raw)

                # ── Phase 1 validation — before caching ──────────────────────

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

def __init__(self):

            super().__init__('T04', 'irs_bmf', 604800, 86400)

        async def fetch(self) -> bytes:

            async with httpx.AsyncClient(timeout=30.0) as client:

                resp = await client.get(IRS_BMF_URL)

                resp.raise_for_status()

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

Raise any exception on failure — run_forever will catch it.

        """

        raise NotImplementedError(

            f"{self.__class__.__name__}.fetch() must be overridden."

        )

    async def run_forever(self) -> None:

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

GET /ops/api/metrics        — JSON metrics snapshot (same data, for fetch())

  GET /ops/health             — {"ok": true}

The dashboard auto-refreshes every 15 seconds via JavaScript fetch().

"""

import logging

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

schedule_seconds=604800,

        )

    async def fetch(self) -> bytes:

        """Fetch 990 filing index and cache financial records per EIN."""

        indexed = 0

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

async function refresh() {

  document.getElementById('refresh-label').textContent = 'fetching…';

  try {

    const res = await fetch('/ops/api/metrics');

    if (!res.ok) throw new Error('HTTP ' + res.status);

    const d = await res.json();

    const t = d.totals || {};

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

def error_response(

    error_code: str,

    message: str,

    query_hash: str = "",

    retry_after: int = 0,

    ingest_healthy: bool = False,

    upstream: str = "",

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

"name": "report_feedback",

      "description": "Report an incorrect, incomplete or stale DataNexus tool result.",

      "parameters": {

        "query_hash": { "type": "string", "description": "16-char hex query hash from the tool response", "required": true },

        "issue": { "type": "string", "description": "Description of the issue", "required": true }

      }

    },

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

"name": "search_datanexus_tools",

      "description": "Search DataNexus tools by describing your task in plain English. Returns the most relevant tools for your query. Token-efficient.",

      "parameters": {

        "query": { "type": "string", "description": "Natural language description of your task", "required": true }

      }

    },

    {

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

# --- from FeedbackInput ---

    tool_id:        str

    query_hash:     str

    signal:         str

    comment:        str = ""

    missing_fields: Optional[List[str]] = None

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

max_length=10,

        description="Tool ID — must be in FEEDBACK_ENABLED_TOOLS.",

    )

    query_hash: str = Field(

        min_length=16,

        max_length=64,

        description="Hash returned by the tool call — links feedback to the audit record.",

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

"""

datanexus/tools/security_stateful.py — Sprint 6 stateful security tools.

Tools:

  fetch_cve_watch     — persistent CVE watchlist (create/check/delete)

  audit_sbom_continuous — continuous SBOM monitoring (register/check/deregister)

Redis key schema (dn: prefix):

  dn:cve_watch:{watch_id}    — Hash (watch data)

  dn:cve_watch_ids           — SET  (watch index)

  dn:sbom_watch:{watch_id}   — Hash (SBOM watch data)

  dn:sbom_watch_ids          — SET  (SBOM watch index)

Scheduler: see datanex

Remediation

Add a structured audit-event emit immediately after every destructive sink. Minimum schema: actor, action, target, outcome, request_id. Python: from acme.audit import audit_log @mcp.tool() def delete_record(token: str, record_id: str) -> dict: actor = verify_token(token) db.execute("DELETE FROM records WHERE id = %s", (record_id,)) audit_log( actor=actor.sub, action="delete_record", target=record_id, outcome=

Evidence

"""

datanexus/tools/t11.py — T11 Global Patent Intelligence tool.

Spec: DataNexus_MCP_Spec_v7_4.docx  Section 4, T11 entry

Exactly 4 data functions. Shared infrastructure tools (report_feedback,

report_mcpize_link) are registered ONCE in main.py — NOT here.

Data sources:

  Primary:   EPO OPS (Open Patent Services) — ops.epo.org — OAuth client_credentials

             European Patent Office API; 4 GB/month free tier

  Secondary: USPTO PatentsView — api.patentsview.org — no key required

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _set_redis_client(client: Optional[redis_lib.Redis]) -> None:

    """Inject a Redis client — used in tests (e.g. fakeredis). Pass None to reset."""

    global _redis_client

    _redis_client = client

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _get_frontend_corpus() -> list:

    global _FRONTEND_CORPUS

    if _FRONTEND_CORPUS is None:

        try:

            with open(_CORPUS_PATH) as f:

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _get_redis() -> Optional[redis_lib.Redis]:

    """Lazy Redis connection. Returns None if unavailable — never raises."""

    global _redis_client

    if _redis_client is not None:

        return _redis_client

    try:

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

"""

datanexus/tools/t10.py — T10 OSS Dependency & Vulnerability Intelligence.

Spec: DataNexus_MCP_Spec_v7_3.docx  Section 11.3 / Table 140  (authoritative)

Exactly 5 data functions + 2 infrastructure stubs = 7 total.

Data sources:

  Primary:   Google OSV.dev API (api.osv.dev/v1) — Apache 2.0, no key.

  Secondary: NIST NVD CVE API (services.nvd.nist.gov) — public domain, no key.

  Supporting: deps.dev API (api.deps.dev/v3alpha) — Apache 2.0, no key.

Hard stops (Section 12.5):

  - NEVER return

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _set_redis_client(client: Optional[redis_lib.Redis]) -> None:

    """Inject a Redis client (e.g. fakeredis) for testing. Pass None to reset."""

    global _redis_client

    _redis_client = client

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

"""

datanexus/tools/nonprofit_sprint7.py — Sprint 7 nonprofit depth tools.

Tools:

  search_nonprofits_by_category    — search US nonprofits by NTEE category + state

  fetch_nonprofit_financial_trends — multi-year revenue/expense/asset trends for a nonprofit

OQ1 resolved: ProPublica /api/v2/organizations/{ein}.json returns pre-computed

fields (totrevenue, totfuncexpns, totprgmrevnue, netassetsend, tax_prd_yr) in

filings_with_data. No raw 990 JSON parsing needed.

Circuit breaker: _propublica_br

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

"""

datanexus/tools/security_sprint6.py — Sprint 6 security tools.

Tools:

  fetch_package_maintainer_history  — maintainer health + anomaly score

  fetch_package_risk_brief          — SHIP/CAUTION/BLOCK aggregator

  detect_typosquatting              — Damerau-Levenshtein vs top-10k (added in final step)

All are thin MCP wrappers. Logic lives in _security_utils.py / _maintainer_utils.py.

HTTP self-calls are forbidden — utilities are called directly.

"""

import asyncio

import logging

import tim

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

connection is attempted rather than returning a broken client whose write errors

    would be silently swallowed by the outer try/except in report_feedback().

    """

    global _redis_client

    if _redis_client is not None:

        try:

            _redis_client.ping()

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _set_caller_id(caller_id: Optional[str]) -> None:

    """Inject a fixed caller_id for testing. Pass None to reset."""

    global _test_caller_id

    _test_caller_id = caller_id

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

"""

datanexus/tools/t04.py — T04 IRS 990 / Nonprofit Data tool.

Spec: DataNexus_MCP_Spec_v7_3.docx  Section 12.4 / Phase 2 Step B

Exactly 3 data functions + 2 infrastructure stubs = 5 total.

(Section 11.3, Table 163 — authoritative signatures)

Data sources:

  Primary:   IRS EO BMF (irs.gov) — public domain

  Secondary: IRS TEOS bulk downloads (irs.gov) — public domain

  Tertiary:  UK Charity Commission public bulk extract (no auth required)

             ccewuksprdoneregsadata1.blob.core.windo

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _get_redis() -> Optional[redis_lib.Redis]:

    """Lazy Redis connection. Returns None if unavailable — never raises."""

    global _redis_client

    if _redis_client is not None:

        return _redis_client

    try:

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

try:

                r = await get_redis()

                if r:

                    await r.set(cache_key, tier, ex=300)

            except Exception:

                pass

            return tier

        except Exception as exc:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

# remaining allowlist domains — any path counts

        for allowed in _PATCH_ALLOWLIST:

            if host == allowed or host.endswith("." + allowed):

                return True

    except Exception:

        pass

    return False

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

raw = r.hget(key_feedback(tool_id, rid), "data")

                    if raw:

                        try:

                            records.append(json.loads(raw))

                        except Exception:

                            pass

            except Exception:

                pass

        results[tool_id] = records

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

try:

        r = _get_redis()

        if r:

            r.set(LAST_ERROR_KEY, msg[:500], ex=7 * 86400)

    except Exception:

        pass

async def refresh() -> bool:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

try:

                            records.append(json.loads(raw))

                        except Exception:

                            pass

            except Exception:

                pass

        results[tool_id] = records

    return JSONResponse({"feedback": results})

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

pipe = r.pipeline(transaction=False)

    try:

        pipe.execute()

    except Exception:

        pass

    return count

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

for a in awards:

            ag = a.get("agency") or "Unknown"

            try:

                agency_totals[ag] = agency_totals.get(ag, 0.0) + float(a.get("amount") or 0)

            except (TypeError, ValueError):

                pass

        top_agencies = sorted(agency_totals.items(), key=lambda x: x[1], reverse=True)[:5]

        rows = []

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

return purls

    finally:

        try:

            os.unlink(fname)

        except OSError:

            pass

# ── OSV audit ─────────────────────────────────────────────────────────────────

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

result = []

        for raw in raw_list:

            try:

                result.append(json.loads(raw))

            except Exception:

                pass

        return result

    except Exception as exc:

        log.warning("daily_digest: error reading agent gaps — %s", exc)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

"""Flush PostHog queue on server shutdown."""

    if _ph:

        try:

            _ph.shutdown()

        except Exception:

            pass

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

key = f"datanexus:calls:{tool_id}:{today}"

    try:

        r.incr(key)

        r.expire(key, 35 * 86400)

    except Exception:

        pass

# ── EPO OAuth token management ────────────────────────────────────────────────

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

pipe = r.pipeline(transaction=False)

    try:

        pipe.execute()

    except Exception:

        pass

    return count, sample

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

return purls

        finally:

            try:

                os.unlink(fname)

            except OSError:

                pass

    except Exception:

        pass

    # Fallback: manual extraction from SPDX JSON

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

pipe = r.pipeline(transaction=False)

    try:

        pipe.execute()

    except Exception:

        pass

    return count

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if path.startswith("node_modules/"):

                name = path[len("node_modules/"):]

                pinned[name] = info.get("version", "")

        return pinned

    except json.JSONDecodeError:

        pass

    # yarn.lock: simple regex extraction

    pattern = re.compile(r'^"?(.+?)@[^:]+:\n\s+version "([^"]+)"', re.MULTILINE)

    for m in pattern.finditer(lockfile):

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if _failure_classifier is None:

        try:

            from datanexus.core import failure_classifier as _fc

            _failure_classifier = _fc

        except Exception:

            pass

    return _failure_classifier

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

try:

            if is_re_signup and old_key_hash:

                await r.delete(f"dn:apikey:{old_key_hash}")

            await r.set(f"dn:apikey:{key_hash}", "free", ex=_KEY_CACHE_TTL)

        except Exception:

            pass

    # ── 4. Send email via SendGrid ────────────────────────────────────────────

    email_sent = False

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if r is not None:

                            cache_key = f"datanexus:epo:token:{expiry_ts}"

                            try:

                                r.setex(cache_key, expires_in, token)

                            except Exception:

                                pass

                    record_success_sync("epo_ops")

                    token_bytes = resp.content

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if raw:

                    record = json.loads(raw)

                    if name_upper in record.get("name", "").upper():

                        return record

        except Exception:

            pass

    # Slow fallback: bulk extract search

    results = await search_uk_charity_bulk_by_name(name, limit=1)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

key = f"datanexus:calls:{tool_id}:{today}"

    try:

        r.incr(key)

        r.expire(key, 35 * 86400)

    except Exception:

        pass

def _sam_gov_key() -> str:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

key = f"datanexus:calls:{tool_id}:{today}"

    try:

        r.incr(key)

        r.expire(key, 35 * 86400)

    except Exception:

        pass

def _regs_gov_key() -> str:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

result = []

        for raw in raw_list:

            try:

                result.append(json.loads(raw))

            except Exception:

                pass

        return result

    except Exception as exc:

        log.warning("daily_digest: error reading failures — %s", exc)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

data = resp.json()

                answers = data.get("Answer", [])

                if answers:

                    return selector

        except Exception:

            pass

        return None

    tasks = [_check_one(sel) for sel in selectors]

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if not vector:

            return "UNKNOWN"

        try:

            return _score_to_level(float(vector))

        except (ValueError, TypeError):

            pass

        raw = vector.upper()

        for prefix in ("CVSS:3.1/", "CVSS:3.0/", "CVSS:2.0/"):

            if raw.startswith(prefix):

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

# Fast path: bare numeric score (e.g. stored as "7.5")

    try:

        return _score_to_level(float(vector))

    except (ValueError, TypeError):

        pass

    raw = vector.upper()

    for prefix in ("CVSS:3.1/", "CVSS:3.0/", "CVSS:2.0/"):

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

try:

            raw = r.get(f"datanexus:T04:bmf:{ein_clean}")

            if raw:

                return json.loads(raw)

        except Exception:

            pass

    # 2. Live CSV stream fallback (BMF worker not yet run)

    return await _lookup_ein_csv_live(ein_clean)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

count_key = f"datanexus:ht:daily_count:{day_key}"

                    current   = _r.get(count_key)

                    if current is not None and int(current) >= _HT_DAILY_LIMIT:

                        quota_exceeded = True

                except Exception:

                    pass

            if quota_exceeded:

                return error_response(

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

exc_info=exc_info,

                user_agent=user_agent,

                exc_type=exc_type,

            )

    except Exception:

        pass  # classifier must never affect the response

    return result

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

for result in data.get("results", [])[:20]:

                        a = _parse_usaspending_result(result)

                        try:

                            total_amount += float(a.get("amount") or 0)

                        except (TypeError, ValueError):

                            pass

                        awards.append(a)

                    source_used = "USASpending.gov"

                    record_success_sync("usaspending")

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if sig in BUG_SIGNALS:

                            bugs += 1

                        elif sig in IMPROVEMENT_SIGNALS:

                            improvements += 1

                    except Exception:

                        pass

            except Exception:

                pass