mirror of
https://github.com/storytold/craft-libs.git
synced 2026-10-31 07:39:23 +00:00
Merge pull request #229 from storytold/sonar-fix/path-cmd-hardening
security: validate CLI paths/args in dev tools (SonarCloud S2083/S8707/S8705)
This commit is contained in:
1 file changed
+29
-2
@@ -4,13 +4,28 @@
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
|
||||
class _HttpsOnlyRedirectHandler(urllib.request.HTTPRedirectHandler):
|
||||
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
||||
if urllib.parse.urlsplit(newurl).scheme != "https":
|
||||
raise ValueError("refusing to follow a non-HTTPS redirect")
|
||||
return super().redirect_request(req, fp, code, msg, headers, newurl)
|
||||
|
||||
|
||||
urllib.request.install_opener(urllib.request.build_opener(_HttpsOnlyRedirectHandler))
|
||||
|
||||
|
||||
def fetch(url):
|
||||
if urllib.parse.urlsplit(url).scheme != "https":
|
||||
raise ValueError("refusing to fetch a non-HTTPS URL")
|
||||
for attempt in range(3):
|
||||
try:
|
||||
with urllib.request.urlopen(url, timeout=60) as response:
|
||||
@@ -42,14 +57,26 @@ def main():
|
||||
name, url, digest = fixture["name"], fixture["source"], fixture["sha256"]
|
||||
if Path(name).name != name or name in ("", ".", "..") or not url.startswith("https://"):
|
||||
raise ValueError("invalid fixture name or source")
|
||||
path = destination / name
|
||||
path = (destination / name).resolve()
|
||||
if path.parent != destination:
|
||||
raise ValueError(f"fixture path escapes destination: {name}")
|
||||
if path.is_file() and hashlib.sha256(path.read_bytes()).hexdigest() == digest:
|
||||
print(f"Verified {name}")
|
||||
continue
|
||||
data = fetch(url)
|
||||
if len(data) > 16 * 1024 * 1024 or hashlib.sha256(data).hexdigest() != digest:
|
||||
raise ValueError(f"fixture size or SHA-256 mismatch: {name}")
|
||||
path.write_bytes(data)
|
||||
fd, tmp = tempfile.mkstemp(dir=destination, prefix=f".{name}.", suffix=".tmp")
|
||||
try:
|
||||
with os.fdopen(fd, "wb") as handle:
|
||||
handle.write(data)
|
||||
os.replace(tmp, destination / name)
|
||||
except BaseException:
|
||||
try:
|
||||
os.unlink(tmp)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
print(f"Fetched and verified {name}")
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user