@@ -382,12 +382,41 @@ def _cmd_install_model(args: argparse.Namespace) -> int:
382382 present (unless --force on the parent command).
383383 """
384384 import shutil
385- import subprocess
385+ import re
386+ import subprocess # noqa: S404 — used with absolute paths from shutil.which + list args, never shell=True
387+ import urllib .parse
386388 import urllib .request
387389
388390 hf_repo = "star-ga/mind-mem-4b"
391+
392+ # Validate args.model: filename only, no path traversal, no URL injection.
393+ if not re .fullmatch (r"[A-Za-z0-9._-]+\.gguf" , args .model ):
394+ print (json .dumps ({"error" : f"invalid --model { args .model !r} \\ .gguf" }, indent = 2 ))
395+ return 1
396+ # Validate args.name: alphanumerics + : _ . - / — Ollama tag charset.
397+ if not re .fullmatch (r"[A-Za-z0-9._:/\-]+" , args .name ):
398+ print (json .dumps ({"error" : f"invalid --name { args .name !r} }, indent = 2 ))
399+ return 1
400+ # Validate args.keep-alive: -1 | <number><unit> (e.g. 30m, 1h, 24h)
401+ if not re .fullmatch (r"(-1|\d+(s|m|h|d)?)" , str (args .keep_alive )):
402+ print (json .dumps ({"error" : f"invalid --keep-alive { args .keep_alive !r} }, indent = 2 ))
403+ return 1
404+
389405 gguf_url = f"https://huggingface.co/{ hf_repo } { args .model }
390- dest = os .path .expanduser (args .dest )
406+ # Defense in depth: confirm the URL we built is HTTPS + huggingface.co.
407+ parsed = urllib .parse .urlparse (gguf_url )
408+ if parsed .scheme != "https" or parsed .hostname != "huggingface.co" :
409+ print (json .dumps ({"error" : "internal: refusing to fetch from non-HF URL" }, indent = 2 ))
410+ return 1
411+ dest = os .path .realpath (os .path .expanduser (args .dest ))
412+ # Refuse writes to canonical system paths. We allow $HOME-symlinked-to-/data
413+ # (common on workstations with a separate SSD for ~/.cache), so we deny by
414+ # blacklist instead of confining by allowlist.
415+ _SYSTEM_PREFIXES = ("/etc/" , "/usr/" , "/bin/" , "/sbin/" , "/lib/" , "/lib64/" ,
416+ "/var/" , "/sys/" , "/proc/" , "/dev/" , "/root/" , "/boot/" )
417+ if any (dest .startswith (p ) for p in _SYSTEM_PREFIXES ):
418+ print (json .dumps ({"error" : f"refusing to write to system path: { dest } }, indent = 2 ))
419+ return 1
391420
392421 output : dict [str , Any ] = {
393422 "model_file" : args .model ,
@@ -414,11 +443,14 @@ def _cmd_install_model(args: argparse.Namespace) -> int:
414443 print (json .dumps (output , indent = 2 ))
415444 return 2
416445
417- # 2. Download GGUF (skip if dest already correct size)
446+ # 2. Download GGUF (skip if dest already correct size).
447+ # URL is constrained above to https://huggingface.co/<known repo>/<validated filename>;
448+ # bandit B310 doesn't see the validation but it's enforced.
418449 os .makedirs (os .path .dirname (dest ), exist_ok = True )
419450 expected_size = None
451+ req = urllib .request .Request (gguf_url , method = "HEAD" )
420452 try :
421- with urllib .request .urlopen (gguf_url ) as resp :
453+ with urllib .request .urlopen (req , timeout = 30 ) as resp : # noqa: S310 — URL validated above
422454 expected_size = int (resp .headers .get ("Content-Length" ) or 0 )
423455 except Exception as exc :
424456 output ["error" ] = f"could not query HF for { args .model } { exc }
@@ -430,7 +462,10 @@ def _cmd_install_model(args: argparse.Namespace) -> int:
430462 output ["reason" ] = "dest already present with matching size"
431463 else :
432464 try :
433- urllib .request .urlretrieve (gguf_url , dest )
465+ req = urllib .request .Request (gguf_url )
466+ with urllib .request .urlopen (req , timeout = 600 ) as resp , open (dest , "wb" ) as fh : # noqa: S310 — URL validated above
467+ while chunk := resp .read (8 * 1024 * 1024 ):
468+ fh .write (chunk )
434469 output ["downloaded" ] = True
435470 output ["bytes" ] = os .path .getsize (dest )
436471 except Exception as exc :
@@ -451,10 +486,17 @@ def _cmd_install_model(args: argparse.Namespace) -> int:
451486 fh .write (modelfile_body )
452487 output ["modelfile" ] = modelfile
453488
454- # 4. Ollama import
489+ # 4. Ollama import — args.name and modelfile are validated above;
490+ # we resolve `ollama` to its absolute path and pass argv as a list
491+ # (never shell=True) so B603/B607 do not apply.
492+ ollama_bin = shutil .which ("ollama" )
493+ if not ollama_bin :
494+ output ["error" ] = "ollama disappeared from PATH between checks"
495+ print (json .dumps (output , indent = 2 ))
496+ return 2
455497 try :
456- result = subprocess .run (
457- ["ollama" , "create" , args .name , "-f" , modelfile ],
498+ result = subprocess .run ( # noqa: S603 — argv list, no shell, validated args
499+ [ollama_bin , "create" , args .name , "-f" , modelfile ],
458500 capture_output = True ,
459501 text = True ,
460502 timeout = 180 ,
@@ -470,10 +512,11 @@ def _cmd_install_model(args: argparse.Namespace) -> int:
470512 print (json .dumps (output , indent = 2 ))
471513 return 6
472514
473- # 5. Smoke test (warm the model + keep-alive)
515+ # 5. Smoke test (warm the model + keep-alive). Same safety profile
516+ # as step 4: absolute path + argv list + validated args, no shell.
474517 try :
475- smoke = subprocess .run (
476- ["ollama" , "run" , args .name , "test" ],
518+ smoke = subprocess .run ( # noqa: S603 — argv list, no shell, validated args
519+ [ollama_bin , "run" , args .name , "test" ],
477520 input = "hi\n " ,
478521 capture_output = True ,
479522 text = True ,
0 commit comments