|
| 1 | +"""An ad-hoc code source for the whitebox read. |
| 2 | +
|
| 3 | +When the code behind a site is not one of the scanned repositories, a person can |
| 4 | +point the scan at the code directly: a public git URL to clone, or an uploaded |
| 5 | +zip or tar.gz. The routes and sinks read from it then guide the dynamic scan |
| 6 | +the same way a mapped repository's would. |
| 7 | +
|
| 8 | +Both are read only and handled with care. The clone is refused unless it is a |
| 9 | +plain https URL to a public host, so it cannot be turned into a request against |
| 10 | +the network the scanner sits in, and it never prompts for or stores a |
| 11 | +credential. The archive is unpacked with hard limits on size and count, and |
| 12 | +every entry is checked so nothing can be written outside its own folder. |
| 13 | +""" |
| 14 | + |
| 15 | +from __future__ import annotations |
| 16 | + |
| 17 | +import contextlib |
| 18 | +import os |
| 19 | +import shutil |
| 20 | +import subprocess |
| 21 | +import tarfile |
| 22 | +import tempfile |
| 23 | +import zipfile |
| 24 | +from pathlib import Path |
| 25 | + |
| 26 | +MAX_ARCHIVE_BYTES = 200 * 1024 * 1024 # the stored upload |
| 27 | +MAX_UNCOMPRESSED = 1024 * 1024 * 1024 # what it unpacks to |
| 28 | +MAX_ENTRIES = 50000 |
| 29 | +CLONE_TIMEOUT = 180 |
| 30 | +_EXT = {"zip": ".zip", "tgz": ".tar.gz"} |
| 31 | + |
| 32 | + |
| 33 | +def _adhoc_dir() -> Path: |
| 34 | + from app.core.external import data_dir |
| 35 | + |
| 36 | + d = data_dir() / "adhoc" |
| 37 | + d.mkdir(parents=True, exist_ok=True) |
| 38 | + return d |
| 39 | + |
| 40 | + |
| 41 | +def archive_path(asset_id, kind: str) -> Path: |
| 42 | + return _adhoc_dir() / f"{int(asset_id)}{_EXT.get(kind, '.bin')}" |
| 43 | + |
| 44 | + |
| 45 | +def clear_archive(asset_id) -> None: |
| 46 | + for ext in _EXT.values(): |
| 47 | + (_adhoc_dir() / f"{int(asset_id)}{ext}").unlink(missing_ok=True) |
| 48 | + |
| 49 | + |
| 50 | +def kind_of(filename: str) -> str: |
| 51 | + n = (filename or "").lower() |
| 52 | + if n.endswith(".zip"): |
| 53 | + return "zip" |
| 54 | + if n.endswith(".tar.gz") or n.endswith(".tgz"): |
| 55 | + return "tgz" |
| 56 | + return "" |
| 57 | + |
| 58 | + |
| 59 | +async def save_upload(asset_id, upload) -> str: |
| 60 | + """Store an uploaded zip or tar.gz for a site. -> its kind, or raises |
| 61 | + ValueError with a sentence to show. Streamed, with a size cap, so a huge |
| 62 | + upload cannot fill the disk.""" |
| 63 | + kind = kind_of(getattr(upload, "filename", "")) |
| 64 | + if not kind: |
| 65 | + raise ValueError("The code archive has to be a .zip or a .tar.gz file.") |
| 66 | + clear_archive(asset_id) |
| 67 | + dest = archive_path(asset_id, kind) |
| 68 | + size = 0 |
| 69 | + try: |
| 70 | + with open(dest, "wb") as out: |
| 71 | + while True: |
| 72 | + chunk = await upload.read(1024 * 1024) |
| 73 | + if not chunk: |
| 74 | + break |
| 75 | + size += len(chunk) |
| 76 | + if size > MAX_ARCHIVE_BYTES: |
| 77 | + raise ValueError("The code archive is too large (over 200 MB).") |
| 78 | + out.write(chunk) |
| 79 | + except ValueError: |
| 80 | + dest.unlink(missing_ok=True) |
| 81 | + raise |
| 82 | + if size == 0: |
| 83 | + dest.unlink(missing_ok=True) |
| 84 | + raise ValueError("The code archive was empty.") |
| 85 | + return kind |
| 86 | + |
| 87 | + |
| 88 | +def validate_git_url(url: str) -> str: |
| 89 | + """-> a clean https git URL, or raises ValueError. Refuses anything that is |
| 90 | + not https to a public host, and any URL carrying a credential.""" |
| 91 | + from urllib.parse import urlparse |
| 92 | + |
| 93 | + from app.core import external_probe |
| 94 | + |
| 95 | + url = (url or "").strip() |
| 96 | + if not url: |
| 97 | + return "" |
| 98 | + parsed = urlparse(url) |
| 99 | + if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: |
| 100 | + raise ValueError("The git URL has to be an https address with no user name " |
| 101 | + "or password in it. For private code, upload an archive.") |
| 102 | + if not external_probe._host_is_public(parsed.hostname): |
| 103 | + raise ValueError("The git URL has to be a public host, not a private " |
| 104 | + "address. For internal code, upload an archive.") |
| 105 | + return url |
| 106 | + |
| 107 | + |
| 108 | +def describe(asset) -> dict: |
| 109 | + """What the page may show about the ad-hoc source. {} when there is none.""" |
| 110 | + git = getattr(asset, "adhoc_git_url", "") or "" |
| 111 | + kind = getattr(asset, "adhoc_archive", "") or "" |
| 112 | + if not git and not kind: |
| 113 | + return {} |
| 114 | + return {"git_url": git, "archive_kind": kind, |
| 115 | + "label": git or (f"uploaded {kind} archive" if kind else "")} |
| 116 | + |
| 117 | + |
| 118 | +def _safe_clone(url: str, dest: Path) -> str: |
| 119 | + """Shallow clone url into dest. -> "" on success, else a short reason.""" |
| 120 | + env = dict(os.environ, GIT_TERMINAL_PROMPT="0", GIT_ASKPASS="/bin/true", |
| 121 | + GCM_INTERACTIVE="never") |
| 122 | + cmd = ["git", "-c", "protocol.file.allow=never", "-c", "protocol.ext.allow=never", |
| 123 | + "-c", "core.askpass=/bin/true", "clone", "--depth", "1", |
| 124 | + "--single-branch", "--no-tags", url, str(dest)] |
| 125 | + try: |
| 126 | + r = subprocess.run(cmd, env=env, capture_output=True, text=True, |
| 127 | + timeout=CLONE_TIMEOUT) |
| 128 | + except subprocess.TimeoutExpired: |
| 129 | + return "the clone took too long and was stopped." |
| 130 | + if r.returncode != 0: |
| 131 | + return f"git clone failed: {(r.stderr or '').strip().splitlines()[-1][:200] if r.stderr else 'unknown error'}" |
| 132 | + return "" |
| 133 | + |
| 134 | + |
| 135 | +def _safe_zip(path: Path, dest: Path) -> str: |
| 136 | + dest_root = str(dest.resolve()) |
| 137 | + total = 0 |
| 138 | + with zipfile.ZipFile(path) as z: |
| 139 | + infos = z.infolist() |
| 140 | + if len(infos) > MAX_ENTRIES: |
| 141 | + return "the archive has too many files in it." |
| 142 | + for info in infos: |
| 143 | + if info.is_dir(): |
| 144 | + continue |
| 145 | + target = (dest / info.filename).resolve() |
| 146 | + if target != dest and not str(target).startswith(dest_root + os.sep): |
| 147 | + return "the archive tries to write outside its own folder." |
| 148 | + total += info.file_size |
| 149 | + if total > MAX_UNCOMPRESSED: |
| 150 | + return "the archive is too large when it is unpacked." |
| 151 | + target.parent.mkdir(parents=True, exist_ok=True) |
| 152 | + with z.open(info) as src, open(target, "wb") as out: |
| 153 | + shutil.copyfileobj(src, out, 1024 * 1024) |
| 154 | + return "" |
| 155 | + |
| 156 | + |
| 157 | +def _safe_tar(path: Path, dest: Path) -> str: |
| 158 | + with tarfile.open(path, "r:gz") as t: |
| 159 | + members = t.getmembers() |
| 160 | + if len(members) > MAX_ENTRIES: |
| 161 | + return "the archive has too many files in it." |
| 162 | + if sum(m.size for m in members if m.isreg()) > MAX_UNCOMPRESSED: |
| 163 | + return "the archive is too large when it is unpacked." |
| 164 | + # The data filter (Python 3.12+) blocks absolute paths, traversal and |
| 165 | + # links that point outside the destination. |
| 166 | + t.extractall(dest, filter="data") |
| 167 | + return "" |
| 168 | + |
| 169 | + |
| 170 | +@contextlib.contextmanager |
| 171 | +def checkout(adhoc: dict): |
| 172 | + """Materialise the ad-hoc source into a temporary folder, cleaned up on exit. |
| 173 | +
|
| 174 | + adhoc is {git, archive, asset_id}. Yields (root, why): root is the folder |
| 175 | + to read, or None with a short reason when there is nothing set or it failed. |
| 176 | + """ |
| 177 | + from app.core import recheck |
| 178 | + |
| 179 | + git = (adhoc.get("git") or "").strip() |
| 180 | + kind = (adhoc.get("archive") or "").strip() |
| 181 | + asset_id = adhoc.get("asset_id") |
| 182 | + if not git and not kind: |
| 183 | + yield None, "" |
| 184 | + return |
| 185 | + base = recheck.memory_dir() or None |
| 186 | + tmp = tempfile.mkdtemp(prefix="gcr-adhoc-", dir=base) |
| 187 | + try: |
| 188 | + dest = Path(tmp) / "src" |
| 189 | + if git: |
| 190 | + try: |
| 191 | + url = validate_git_url(git) |
| 192 | + except ValueError as exc: |
| 193 | + yield None, str(exc) |
| 194 | + return |
| 195 | + why = _safe_clone(url, dest) |
| 196 | + else: |
| 197 | + dest.mkdir(parents=True, exist_ok=True) |
| 198 | + archive = archive_path(asset_id, kind) |
| 199 | + if not archive.exists(): |
| 200 | + yield None, "the uploaded archive is missing." |
| 201 | + return |
| 202 | + why = _safe_zip(archive, dest) if kind == "zip" else _safe_tar(archive, dest) |
| 203 | + if why: |
| 204 | + yield None, why |
| 205 | + return |
| 206 | + yield str(dest), "" |
| 207 | + finally: |
| 208 | + shutil.rmtree(tmp, ignore_errors=True) |
0 commit comments