diff --git a/backend/util/cl2k/gdrive_upload.py b/backend/util/cl2k/gdrive_upload.py index 854edaf6..b272a71c 100644 --- a/backend/util/cl2k/gdrive_upload.py +++ b/backend/util/cl2k/gdrive_upload.py @@ -14,10 +14,22 @@ import shutil import subprocess import tempfile +import threading import uuid +from collections import defaultdict from shutil import which from typing import Any, List, Optional +# Serialises writes per Drive folder: concurrent uploads both list "absent" and +# both create, and Drive allows duplicate names. Thread lock — single process only. +_FOLDER_LOCKS_GUARD = threading.Lock() +_FOLDER_LOCKS: "defaultdict[str, threading.Lock]" = defaultdict(threading.Lock) + + +def _folder_lock(folder_id: str) -> threading.Lock: + with _FOLDER_LOCKS_GUARD: + return _FOLDER_LOCKS[folder_id] + def _rclone_path() -> str: env = os.getenv("RCLONE_PATH") @@ -88,11 +100,18 @@ def _oauth_args(sync_cfg: Any) -> List[str]: token = (token or "").strip() if "access_token" not in token and "refresh_token" not in token: return [] + client_id = getattr(sync_cfg, "client_id", "") or "" + client_secret = getattr(sync_cfg, "client_secret", "") or "" + # A leading "-" would reach rclone as an option, not data. Real Google + # credentials never start with one. + _reject_unsafe(client_id, "gdrive_client_id") + _reject_unsafe(client_secret, "gdrive_client_secret") + _reject_unsafe(token, "gdrive_token") return [ "--drive-client-id", - getattr(sync_cfg, "client_id", "") or "", + client_id, "--drive-client-secret", - getattr(sync_cfg, "client_secret", "") or "", + client_secret, "--drive-token", token, ] @@ -143,6 +162,7 @@ def upload_file( non-zero rclone exit so the caller can record the failure. """ _reject_unsafe_id(folder_id, "gdrive_folder_id") + _reject_unsafe(local_path, "local_path") auth = _upload_auth_args(sync_cfg) if not auth: raise RuntimeError( @@ -163,12 +183,72 @@ def upload_file( "-v", *auth, ] - result = subprocess.run(cmd, check=False, capture_output=True, text=True) - if result.returncode != 0: - raise RuntimeError(f"rclone copy failed: {_rclone_error_detail(result.stderr)}") + # Serialised per folder: see _FOLDER_LOCKS. rclone replaces a same-named file + # correctly on its own — it only duplicates when two runs overlap. + with _folder_lock(folder_id): + result = subprocess.run(cmd, check=False, capture_output=True, text=True) + if result.returncode != 0: + raise RuntimeError( + f"rclone copy failed: {_rclone_error_detail(result.stderr)}" + ) + _reap_duplicates(os.path.basename(local_path), folder_id, sync_cfg, logger) logger.debug(f"uploaded {os.path.basename(local_path)} to drive {folder_id}") +# rclone filter metacharacters. CL2K names carry `{tmdb-…}`, and `{}` is +# alternation — unescaped it matches the wrong thing. +_FILTER_META = re.compile(r"([\\*?\[\]{}])") + + +def _filter_literal(name: str) -> str: + """``name`` as an rclone filter that matches it and nothing else.""" + return "/" + _FILTER_META.sub(r"\\\1", name) + + +def _reap_duplicates(name: str, folder_id: str, sync_cfg: Any, logger) -> None: + """Collapse same-named copies of ``name``, keeping the newest — ours. + + Scoped to ``name``: an upload of one file must never delete another file's + duplicates. Only runs when a duplicate of what we just wrote exists, so the + normal path issues no destructive command. Never raises — the upload + succeeded, and failing to tidy up must not report it as failed. + """ + try: + names = list_files(folder_id, sync_cfg, logger, strict=True) + if sum(1 for n in names if n == name) < 2: + return + logger.warning( + f"CL2K drive {folder_id}: {name} exists more than once — keeping the newest" + ) + auth = _upload_auth_args(sync_cfg) + rclone = _rclone_path() + result = subprocess.run( + [ + rclone, + "dedupe", + "--dedupe-mode", + "newest", + "posters:", + "--drive-root-folder-id", + folder_id, + "--drive-use-trash=false", + # dedupe honours filters from rclone 1.61. + "--include", + _filter_literal(name), + *auth, + ], + check=False, + capture_output=True, + text=True, + ) + if result.returncode != 0: + logger.warning( + f"CL2K drive dedupe failed: {_rclone_error_detail(result.stderr)}" + ) + except Exception as exc: + logger.warning(f"CL2K drive duplicate check failed for {name}: {exc}") + + def move_file( old_name: str, new_name: str, @@ -207,7 +287,10 @@ def move_file( "-v", *auth, ] - result = subprocess.run(cmd, check=False, capture_output=True, text=True) + # Upload's lock: this is the folder's other writer, and a rename onto a name + # an upload is creating duplicates the same way. + with _folder_lock(folder_id): + result = subprocess.run(cmd, check=False, capture_output=True, text=True) if result.returncode != 0: raise RuntimeError( f"rclone moveto failed: {_rclone_error_detail(result.stderr)}" diff --git a/tests/test_cl2k_gdrive_upload_race.py b/tests/test_cl2k_gdrive_upload_race.py new file mode 100644 index 00000000..9d4b3a1b --- /dev/null +++ b/tests/test_cl2k_gdrive_upload_race.py @@ -0,0 +1,255 @@ +"""Concurrent uploads to one Drive folder must not create duplicate names. + +rclone decides create-vs-replace by listing the destination first, and Drive +permits two files sharing a name — so two overlapping uploads both see "absent" +and both create. Reproduced against the real Drive before this fix: sequential +uploads left 1 copy, concurrent uploads left 2. + +The fake below models exactly that: a listing, then a create-or-replace, with a +window between them. +""" + +import threading +import time +import types + +import pytest + +from backend.util.cl2k import gdrive_upload as gd + + +class FakeDrive: + """A folder where create-vs-replace is decided by a listing, with a real + gap between the check and the write — the shape that duplicates.""" + + def __init__(self, gap=0.05): + self.files = [] # names; a list, not a set — Drive allows duplicates + self.gap = gap + self._mutex = threading.Lock() + + def rclone_copy(self, name): + present = name in self.files # check + time.sleep(self.gap) # ...window... + with self._mutex: + if not present: + self.files.append(name) # create (duplicates if two got here) + + def count(self, name): + return sum(1 for n in self.files if n == name) + + +def _install(monkeypatch, drive, name="poster.png"): + """Point upload_file's subprocess at the fake, keeping the real locking.""" + monkeypatch.setattr(gd, "_rclone_path", lambda: "/usr/bin/rclone") + monkeypatch.setattr(gd, "_ensure_remote", lambda _r: None) + monkeypatch.setattr(gd, "_upload_auth_args", lambda _c: ["--drive-token", "x"]) + monkeypatch.setattr( + gd, "list_files", lambda fid, cfg, log, strict=False: list(drive.files) + ) + + def fake_run(cmd, **kw): + if "copy" in cmd: + drive.rclone_copy(name) + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + monkeypatch.setattr(gd.subprocess, "run", fake_run) + + +def _logger(): + noop = lambda *a, **k: None # noqa: E731 + return types.SimpleNamespace(debug=noop, info=noop, warning=noop, error=noop) + + +def _upload_twice_concurrently(folder="folder-A", path="/tmp/poster.png"): + barrier = threading.Barrier(2) + + def worker(): + barrier.wait() # maximise overlap + gd.upload_file(path, folder, object(), _logger()) + + threads = [threading.Thread(target=worker) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + + +def test_the_fake_reproduces_the_duplicate_without_the_lock(monkeypatch): + """Guard the guard: if this ever passes, the fake stopped modelling the bug + and the test below would prove nothing.""" + drive = FakeDrive() + _install(monkeypatch, drive) + monkeypatch.setattr(gd, "_folder_lock", lambda _f: threading.Lock()) # useless lock + _upload_twice_concurrently() + assert drive.count("poster.png") == 2 + + +def test_concurrent_uploads_to_one_folder_leave_a_single_copy(monkeypatch): + drive = FakeDrive() + _install(monkeypatch, drive) + _upload_twice_concurrently() + assert drive.count("poster.png") == 1 + + +def test_different_folders_still_upload_in_parallel(monkeypatch): + """The lock is per folder, so unrelated destinations must not serialise.""" + assert gd._folder_lock("folder-A") is gd._folder_lock("folder-A") + assert gd._folder_lock("folder-A") is not gd._folder_lock("folder-B") + + +def test_a_preexisting_duplicate_is_reaped(monkeypatch): + drive = FakeDrive() + drive.files = ["poster.png", "poster.png"] # e.g. written before this fix + _install(monkeypatch, drive) + deduped = [] + + def fake_run(cmd, **kw): + if "dedupe" in cmd: + deduped.append(cmd) + drive.files = ["poster.png"] + elif "copy" in cmd: + drive.rclone_copy("poster.png") + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + monkeypatch.setattr(gd.subprocess, "run", fake_run) + gd.upload_file("/tmp/poster.png", "folder-A", object(), _logger()) + assert deduped, "an existing duplicate should trigger a dedupe" + assert "newest" in deduped[0], "must keep the newest, i.e. the copy just uploaded" + assert drive.count("poster.png") == 1 + + +def test_no_duplicate_means_no_destructive_command(monkeypatch): + """The normal path must never run dedupe — it deletes.""" + drive = FakeDrive() + _install(monkeypatch, drive) + seen = [] + + def fake_run(cmd, **kw): + seen.append(cmd) + if "copy" in cmd: + drive.rclone_copy("poster.png") + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + monkeypatch.setattr(gd.subprocess, "run", fake_run) + gd.upload_file("/tmp/poster.png", "folder-A", object(), _logger()) + assert not any("dedupe" in c for c in seen) + + +def test_a_rename_serialises_against_an_upload_to_the_same_folder(monkeypatch): + """move_file is the folder's other writer (the poster healer renames while + the maker uploads), so it has to take the same lock or the convention only + covers half the writers.""" + drive = FakeDrive() + _install(monkeypatch, drive) + order = [] + + def fake_run(cmd, **kw): + op = "moveto" if "moveto" in cmd else "copy" + order.append("%s-start" % op) + time.sleep(0.05) + order.append("%s-end" % op) + if op == "copy": + drive.rclone_copy("poster.png") + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + monkeypatch.setattr(gd.subprocess, "run", fake_run) + barrier = threading.Barrier(2) + + def up(): + barrier.wait() + gd.upload_file("/tmp/poster.png", "folder-A", object(), _logger()) + + def mv(): + barrier.wait() + gd.move_file("old.png", "poster.png", "folder-A", object(), _logger()) + + ts = [threading.Thread(target=up), threading.Thread(target=mv)] + for t in ts: + t.start() + for t in ts: + t.join() + # Never interleaved: each write finishes before the other starts. + assert order[0].endswith("-start") and order[1].endswith("-end") + assert order[1].split("-")[0] == order[0].split("-")[0] + + +def test_a_failed_duplicate_check_does_not_fail_the_upload(monkeypatch): + drive = FakeDrive() + _install(monkeypatch, drive) + monkeypatch.setattr( + gd, "list_files", lambda *a, **k: (_ for _ in ()).throw(RuntimeError("api down")) + ) + gd.upload_file("/tmp/poster.png", "folder-A", object(), _logger()) + assert drive.count("poster.png") == 1 + + +def test_every_value_reaching_the_rclone_argv_is_validated(monkeypatch): + """CodeQL flags this call site as an uncontrolled command line. subprocess is + list-form (no shell), so the residual risk is argument injection: a value + starting with "-" is read by rclone as an option. folder_id and the SA path + were already validated; local_path and the OAuth trio were not.""" + drive = FakeDrive() + _install(monkeypatch, drive) + with pytest.raises(ValueError, match="local_path"): + gd.upload_file("-X=evil", "folder-A", object(), _logger()) + + +def test_an_option_shaped_oauth_value_is_refused(monkeypatch): + cfg = types.SimpleNamespace( + token='{"access_token": "a"}', client_id="-oops", client_secret="" + ) + with pytest.raises(ValueError, match="gdrive_client_id"): + gd._oauth_args(cfg) + + +def test_real_google_credentials_still_pass(monkeypatch): + """The validator must not reject legitimate config: ids are numeric-led, + secrets are GOCSPX-…, and the token is JSON.""" + cfg = types.SimpleNamespace( + token='{"access_token": "ya29.x", "refresh_token": "1//y"}', + client_id="123456789-abc.apps.googleusercontent.com", + client_secret="GOCSPX-abcdef123456", + ) + args = gd._oauth_args(cfg) + assert "--drive-token" in args and "GOCSPX-abcdef123456" in args + + +def test_the_reap_is_scoped_to_the_uploaded_filename(monkeypatch): + """Uploading one file must never collapse another file's duplicates. Proven + against the real Drive too: staging duplicates of two names and reaping one + left the other at 2.""" + drive = FakeDrive() + drive.files = ["poster.png", "poster.png", "other.png", "other.png"] + _install(monkeypatch, drive) + seen = [] + + def fake_run(cmd, **kw): + seen.append(cmd) + if "dedupe" in cmd: + # Actually APPLY the filter rather than assuming it. A fake that + # collapses a hard-coded name passes even when the filter is wrong, + # which proves nothing about the thing under test. + pattern = cmd[cmd.index("--include") + 1] + target = pattern.lstrip("/").replace("\\", "") + while drive.files.count(target) > 1: + drive.files.remove(target) + elif "copy" in cmd: + drive.rclone_copy("poster.png") + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + monkeypatch.setattr(gd.subprocess, "run", fake_run) + gd.upload_file("/tmp/poster.png", "folder-A", object(), _logger()) + + dedupe = [c for c in seen if "dedupe" in c] + assert dedupe, "a duplicate of the uploaded name should trigger a dedupe" + assert dedupe[0][dedupe[0].index("--include") + 1] == "/poster.png" + assert drive.count("poster.png") == 1, "our own duplicate must be collapsed" + assert drive.count("other.png") == 2, "another file's duplicates must survive" + + +def test_filter_escapes_the_metacharacters_in_real_filenames(monkeypatch): + """CL2K names carry `{tmdb-…}`, and `{}` is alternation in an rclone filter — + unescaped, the pattern matches the wrong thing or nothing at all.""" + got = gd._filter_literal("Cassadaga (2011) {tmdb-92398} - logo.png") + assert got == r"/Cassadaga (2011) \{tmdb-92398\} - logo.png" + assert gd._filter_literal("a[b]*c?d") == r"/a\[b\]\*c\?d"