|
35 | 35 | from __future__ import annotations |
36 | 36 |
|
37 | 37 | import argparse |
| 38 | +import re |
38 | 39 | from pathlib import Path |
39 | 40 | from typing import Any |
40 | 41 |
|
@@ -275,23 +276,45 @@ def _write_cif_file(output_path: Path, lines: list[str]) -> None: |
275 | 276 | output_path.write_text("\n".join(lines), encoding="utf-8") |
276 | 277 |
|
277 | 278 |
|
278 | | -def _validated_output_path(output_dir: Path, refcode: str) -> Path: |
279 | | - base_dir = output_dir.expanduser().resolve() |
280 | | - candidate = (base_dir / f"{refcode}.cif").resolve() |
281 | | - try: |
282 | | - candidate.relative_to(base_dir) |
283 | | - except ValueError as exc: |
284 | | - raise ValueError(f"Refusing to write outside output directory: {candidate}") from exc |
285 | | - return candidate |
| 279 | +_SAFE_REFCODE_RE = re.compile(r"^[A-Za-z][A-Za-z0-9]{1,19}$") |
286 | 280 |
|
287 | 281 |
|
288 | | -def export_full_cif_from_csd(refcode: str, output_path: Path) -> dict[str, Any]: |
| 282 | +def _sanitize_refcode(refcode: str) -> str: |
| 283 | + """Ensure refcode contains only safe alphanumeric characters.""" |
| 284 | + if not _SAFE_REFCODE_RE.match(refcode): |
| 285 | + raise ValueError( |
| 286 | + f"Invalid refcode {refcode!r}: must be 2-20 alphanumeric characters starting with a letter" |
| 287 | + ) |
| 288 | + return refcode |
| 289 | + |
| 290 | + |
| 291 | +def _safe_output_dir(raw_path: str) -> Path: |
| 292 | + """Resolve and validate output directory, rejecting path traversal.""" |
| 293 | + # Only allow relative paths under the current working directory |
| 294 | + if raw_path != Path(raw_path).as_posix().replace("\\", "/"): |
| 295 | + cleaned = Path(raw_path).as_posix() |
| 296 | + else: |
| 297 | + cleaned = raw_path |
| 298 | + if ".." in cleaned.split("/"): |
| 299 | + raise ValueError(f"Path traversal not allowed in output directory: {raw_path!r}") |
| 300 | + resolved = Path.cwd() / cleaned |
| 301 | + resolved = resolved.resolve() |
| 302 | + cwd = Path.cwd().resolve() |
| 303 | + if not str(resolved).startswith(str(cwd)): |
| 304 | + raise ValueError(f"Output directory must be under working directory: {resolved}") |
| 305 | + return resolved |
| 306 | + |
| 307 | + |
| 308 | +def export_full_cif_from_csd(refcode: str, output_dir: Path) -> dict[str, Any]: |
289 | 309 | """Export a CSD structure to CIF with occupancy and disorder metadata.""" |
290 | 310 | from ccdc.io import EntryReader |
291 | 311 |
|
| 312 | + safe_name = _sanitize_refcode(refcode) |
| 313 | + output_path = output_dir / f"{safe_name}.cif" |
| 314 | + |
292 | 315 | with EntryReader("CSD") as reader: |
293 | | - entry = reader.entry(refcode) |
294 | | - crystal = reader.crystal(refcode) |
| 316 | + entry = reader.entry(safe_name) |
| 317 | + crystal = reader.crystal(safe_name) |
295 | 318 |
|
296 | 319 | molecule = crystal.disordered_molecule or crystal.molecule |
297 | 320 | disorder_map = _build_disorder_map(crystal) |
@@ -324,12 +347,11 @@ def parse_args() -> argparse.Namespace: |
324 | 347 |
|
325 | 348 | def main() -> None: |
326 | 349 | args = parse_args() |
327 | | - output_dir = Path(args.output_dir) |
| 350 | + output_dir = _safe_output_dir(args.output_dir) |
328 | 351 |
|
329 | 352 | print(f"Licence requirement: {LICENSE_REQUIREMENT}") |
330 | 353 | for refcode in args.refcodes: |
331 | | - output_path = _validated_output_path(output_dir, refcode) |
332 | | - summary = export_full_cif_from_csd(refcode, output_path) |
| 354 | + summary = export_full_cif_from_csd(refcode, output_dir) |
333 | 355 | print( |
334 | 356 | f"Exported {summary['refcode']} -> {summary['output']} " |
335 | 357 | f"(atoms={summary['n_atoms']}, partial_occ={summary['n_partial_occupancy']}, bonds={summary['n_bonds']})" |
|
0 commit comments