|
32 | 32 | SCRIPT_URL = "https://simbad.cds.unistra.fr/simbad/sim-script" |
33 | 33 | TAP_URL = "https://simbad.cds.unistra.fr/simbad/sim-tap/sync" |
34 | 34 |
|
| 35 | +# Characters that could enable script/SQL injection - must not appear in user input |
| 36 | +_FORBIDDEN_IN_NAME = frozenset("\n\r\t'\"\\;<>") |
| 37 | + |
| 38 | + |
| 39 | +def _sanitize_object_name(name: str) -> str: |
| 40 | + """Validate and sanitize object name for script interface (no injection).""" |
| 41 | + if not name or not isinstance(name, str): |
| 42 | + raise ValueError("Object name must be a non-empty string") |
| 43 | + name = " ".join(name.split()) # collapse whitespace |
| 44 | + if len(name) > 128: |
| 45 | + raise ValueError("Object name too long") |
| 46 | + if any(c in _FORBIDDEN_IN_NAME for c in name): |
| 47 | + raise ValueError("Object name contains disallowed characters") |
| 48 | + return name.strip() |
| 49 | + |
| 50 | + |
| 51 | +def _sanitize_adql_string(s: str) -> str: |
| 52 | + """Escape single quotes for safe use in ADQL string literals.""" |
| 53 | + if not s or not isinstance(s, str): |
| 54 | + raise ValueError("Identifier must be a non-empty string") |
| 55 | + if len(s) > 128: |
| 56 | + raise ValueError("Identifier too long") |
| 57 | + # Block newlines, semicolons, backslashes (injection vectors) |
| 58 | + bad = frozenset("\n\r\t\\;<>\"") |
| 59 | + if any(c in bad for c in s): |
| 60 | + raise ValueError("Identifier contains disallowed characters") |
| 61 | + return s.replace("'", "''") # ADQL string literal escape |
| 62 | + |
| 63 | + |
35 | 64 | FORMAT_STRINGS = { |
36 | 65 | "basic": "%IDLIST(1) | %COO(A D;ICRS) | %OTYPE", |
37 | 66 | "detailed": "%IDLIST(1) | %COO(A D;ICRS) | %OTYPE | %SP | %FLUXLIST(V)", |
@@ -98,11 +127,12 @@ def query_object( |
98 | 127 | Returns: |
99 | 128 | List of result dicts with keys like main_id, coordinates, object_type, etc. |
100 | 129 | """ |
| 130 | + safe_name = _sanitize_object_name(name) |
101 | 131 | fmt = FORMAT_STRINGS.get(output_format, FORMAT_STRINGS["basic"]) |
102 | 132 | script = "\n".join([ |
103 | 133 | "output console=off script=off", |
104 | 134 | f'format object "{fmt}"', |
105 | | - f"query id {name}", |
| 135 | + f"query id {safe_name}", |
106 | 136 | ]) |
107 | 137 | text = _execute_script(script) |
108 | 138 | return _parse_script_response(text) |
@@ -154,11 +184,12 @@ def query_identifiers( |
154 | 184 | Returns: |
155 | 185 | List of result dicts |
156 | 186 | """ |
| 187 | + safe_pattern = _sanitize_object_name(pattern) |
157 | 188 | fmt = FORMAT_STRINGS.get(output_format, FORMAT_STRINGS["basic"]) |
158 | 189 | script = "\n".join([ |
159 | 190 | "output console=off script=off", |
160 | 191 | f'format object "{fmt}"', |
161 | | - f"query id wildcard {pattern}", |
| 192 | + f"query id wildcard {safe_pattern}", |
162 | 193 | ]) |
163 | 194 | text = _execute_script(script) |
164 | 195 | return _parse_script_response(text, max_results=max_results) |
@@ -208,10 +239,11 @@ def get_all_identifiers(name: str) -> List[str]: |
208 | 239 | Returns: |
209 | 240 | List of identifier strings |
210 | 241 | """ |
| 242 | + safe_name = _sanitize_adql_string(name) |
211 | 243 | result = tap_query( |
212 | 244 | f"SELECT i.id FROM ident AS i " |
213 | 245 | f"JOIN basic AS b ON i.oidref = b.oid " |
214 | | - f"WHERE b.main_id = '{name}'", |
| 246 | + f"WHERE b.main_id = '{safe_name}'", |
215 | 247 | max_results=500, |
216 | 248 | fmt="json", |
217 | 249 | ) |
|
0 commit comments