T09 · Insecure Skill Coding Practices
Warning
- Location
- scripts/patent_search_plan.py:198
- Finding
- Mutable Query Plan Allows Queries Against Unapproved BigQuery Tables<![CDATA[ ## Vulnerability Details **File Location**: `scripts/patent_search_plan.py:198-210`, `scripts/patent_search_plan.py:369-374`; `schemas/query_plan.schema.json:9` **Vulnerability Type**: Unrestricted SQL identifier controlled through a query-plan file **Risk Level**: Medium ### Vulnerable Code ```python where_sql = "\n AND ".join(where_parts) match_score_sql = " + ".join(score_terms) if score_terms else "0" sql = f""" SELECT publication_number, country_code, (SELECT text FROM UNNEST(title_localized) WHERE text IS NOT NULL LIMIT 1) AS title, (SELECT text FROM UNNEST(abstract_localized) WHERE text IS NOT NULL LIMIT 1) AS abstract, SUBSTR((SELECT text FROM UNNEST(claims_localized) WHERE text IS NOT NULL LIMIT 1), 1, 1200) AS claims, ARRAY(SELECT name FROM UNNEST(inventor_harmonized)) AS inventors, ARRAY(SELECT name FROM UNNEST(assignee_harmonized)) AS assignees, ARRAY(SELECT code FROM UNNEST(ipc)) AS ipc_codes, ARRAY(SELECT code FROM UNNEST(cpc)) AS cpc_codes, filing_date, publication_date, ({match_score_sql}) AS match_score FROM `{table}` WHERE {where_sql} ORDER BY match_score DESC, publication_date DESC LIMIT @limit """ job_config = bigquery.QueryJobConfig(query_parameters=params) ``` The value is read directly from the mutable plan: ```python table = str(plan.get("table") or "").strip() if not table: raise SystemExit("query_plan.table 为空") policy = _effective_policy(plan, args) client = bigquery.Client() ``` It is subsequently passed into the query builder: ```python sql, job_config = _build_round_query(table=table, round_cfg=round_cfg) ``` The corresponding schema imposes no allowlist or constant restriction: ```json "table": {"type": "string"} ``` ### Technical Analysis Search terms, dates, countries, and other filter values are correctly passed through BigQuery query parameters. However, the table identifier cannot be represented by a normal query parameter and is instead interpolated directly into the SQL stat ...[truncated 2095 chars]
- Remediation
- <![CDATA[ ## Remediation Suggestions 1. Remove the table name from the externally mutable plan and use a code-level constant: ```python PATENTS_TABLE = "patents-public-data.patents.publications" ``` 2. If the field must remain for compatibility, require exact equality before creating the client or running any query: ```python ALLOWED_TABLE = "patents-public-data.patents.publications" table = str(plan.get("table") or "").strip() if table != ALLOWED_TABLE: raise SystemExit(f"Unsupported BigQuery table: {table!r}") ``` 3. Change the schema to a constant or single-value enumeration: ```json "table": { "type": "string", "enum": ["patents-public-data.patents.publications"] } ``` 4. Validate the query plan against `schemas/query_plan.schema.json` inside the executor before using any plan values. Do not rely on a separately documented validation command. 5. Configure the Google identity with least-privilege BigQuery permissions. Avoid granting broad access to unrelated datasets. 6. Add tests proving that alternate table names, malformed identifiers, missing fields, and additional unexpected plan properties are rejected before a network request occurs. ]]>