diff --git a/l4d2web/services/overlay_builders.py b/l4d2web/services/overlay_builders.py index b97c814..c73c903 100644 --- a/l4d2web/services/overlay_builders.py +++ b/l4d2web/services/overlay_builders.py @@ -173,7 +173,7 @@ class WorkshopBuilder: downloaded = 0 cached = 0 - skipped: list[str] = [] + skipped: set[str] = set() # Download phase. for ( @@ -188,7 +188,7 @@ class WorkshopBuilder: f"workshop item {steam_id} skipped: no file_url " f"(steam result: {last_error or 'unknown'})" ) - skipped.append(steam_id) + skipped.add(steam_id) continue target = cache_path(steam_id) needs_download = ( @@ -200,6 +200,8 @@ class WorkshopBuilder: if not needs_download: cached += 1 continue + # download_to_cache only reads steam_id, file_url, file_size, time_updated; + # consumer_app_id and result are required by the dataclass but unused here. meta = WorkshopMetadata( steam_id=steam_id, title=title, @@ -295,7 +297,21 @@ class WorkshopBuilder: return if name in post_removal_existing: continue - os.symlink(target, addons_dir / name) + link_path = addons_dir / name + # Defensive: if a non-symlink file collides with our name, leave it. + if link_path.exists() and not link_path.is_symlink(): + on_stderr( + f"refusing to overwrite non-symlink at {link_path}; manual intervention required" + ) + continue + if link_path.is_symlink(): + # An obsolete symlink not in `existing` (target outside cache). + # We don't manage these — leave alone. + on_stderr( + f"refusing to overwrite foreign symlink at {link_path}" + ) + continue + os.symlink(target, link_path) created += 1 on_stdout( diff --git a/l4d2web/tests/test_overlay_builders.py b/l4d2web/tests/test_overlay_builders.py index fee321f..ad15f7a 100644 --- a/l4d2web/tests/test_overlay_builders.py +++ b/l4d2web/tests/test_overlay_builders.py @@ -693,3 +693,66 @@ def test_workshop_build_cancels_cleanly_during_download_phase(env, tmp_path, mon overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: cancel_flag["v"], ) + + +def test_workshop_build_refuses_to_overwrite_non_symlink_file(env, tmp_path, monkeypatch): + """If a plain file collides with a workshop symlink name, the build logs + a refusal and leaves the file alone instead of crashing.""" + monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path)) + cache_root = tmp_path / "workshop_cache" + user_id, overlay_id = _create_user_and_overlay("ws", "workshop") + item_id = _add_workshop_item("3001", downloaded=True, cache_root=cache_root) + _associate(overlay_id, item_id) + + # Pre-create a plain file (not a symlink) where the builder would place its symlink. + addons = tmp_path / "overlays" / "7" / "left4dead2" / "addons" + addons.mkdir(parents=True, exist_ok=True) + (addons / "3001.vpk").write_bytes(b"manual file, don't touch") + + out, err, on_stdout, on_stderr = _capture_logs() + with session_scope() as s: + from sqlalchemy import select as _sel + overlay = s.scalar(_sel(Overlay).where(Overlay.id == overlay_id)) + s.expunge(overlay) + overlay_builders.WorkshopBuilder().build( + overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False, + ) + + # The plain file is still there, unchanged. + assert (addons / "3001.vpk").exists() + assert not (addons / "3001.vpk").is_symlink() + assert (addons / "3001.vpk").read_bytes() == b"manual file, don't touch" + # And a refusal message was logged. + assert any("refusing to overwrite non-symlink" in line for line in err) + + +def test_workshop_build_refuses_to_overwrite_foreign_symlink(env, tmp_path, monkeypatch): + """A symlink pointing outside the workshop cache is left alone — not + overwritten, not failed.""" + monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path)) + cache_root = tmp_path / "workshop_cache" + user_id, overlay_id = _create_user_and_overlay("ws", "workshop") + item_id = _add_workshop_item("3002", downloaded=True, cache_root=cache_root) + _associate(overlay_id, item_id) + + # Pre-create a symlink pointing outside the cache root. + foreign_target = tmp_path / "elsewhere" / "thing.vpk" + foreign_target.parent.mkdir(parents=True, exist_ok=True) + foreign_target.write_bytes(b"some other vpk") + addons = tmp_path / "overlays" / "7" / "left4dead2" / "addons" + addons.mkdir(parents=True, exist_ok=True) + os.symlink(foreign_target, addons / "3002.vpk") + + out, err, on_stdout, on_stderr = _capture_logs() + with session_scope() as s: + from sqlalchemy import select as _sel + overlay = s.scalar(_sel(Overlay).where(Overlay.id == overlay_id)) + s.expunge(overlay) + overlay_builders.WorkshopBuilder().build( + overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False, + ) + + # The foreign symlink still points where it did. + assert (addons / "3002.vpk").is_symlink() + assert os.readlink(addons / "3002.vpk") == str(foreign_target) + assert any("refusing to overwrite foreign symlink" in line for line in err)