diff --git a/.gitignore b/.gitignore index 5cea5cf77..ae347f3cd 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ **/agents.md agents.md htmlcov/ +federation-shared.env diff --git a/federation-shared.example.env b/federation-shared.example.env new file mode 100644 index 000000000..0578122de --- /dev/null +++ b/federation-shared.example.env @@ -0,0 +1,7 @@ +# Shared by gateway + federation sync (repo root; gateway/scripts/generate-secrets.sh). +# Compose: env_file → ../../federation-shared.env (gateway) or ../federation-shared.env (federation). +# FEDERATION_SYNC_DRF_TOKEN: bootstrap Token for get-federation-sync-api-key. +# FEDERATION_SYNC_SERVER_API_KEY: export Api-Key (sync service writes after mint). + +FEDERATION_SYNC_DRF_TOKEN= +FEDERATION_SYNC_SERVER_API_KEY= diff --git a/gateway/.envs/example/django.env b/gateway/.envs/example/django.env index 1287a26fa..500d77454 100644 --- a/gateway/.envs/example/django.env +++ b/gateway/.envs/example/django.env @@ -8,6 +8,24 @@ HOSTNAME=localhost:8000 API_VERSION=v1 API_KEY= +# FEDERATION +# ------------------------------------------------------------------------------ +# Load federation-shared.env from repo root on gateway + sync compose stacks (generate-secrets). +# prepare_gateway runs init_federation_sync_token (DB Token from FEDERATION_SYNC_DRF_TOKEN). +# Sync startup mints FEDERATION_SYNC_SERVER_API_KEY via get-federation-sync-api-key. +# Set FEDERATION_SITE_NAME (e.g. crc) when enabling federation; use SDS_SITE_FQDN for the public host (RFC [site].fqdn). +# Master switch for export APIs and Redis federation events. +# FEDERATION_ENABLED=true +# RFC [site].name (short peer id); set SDS_SITE_FQDN separately for [site].fqdn. +# FEDERATION_SITE_NAME=crc +# Redis pub/sub channel federation-sync subscribes to. +# FEDERATION_EVENTS_CHANNEL=federation:events +# Health probe target (federation-sync service). +# FEDERATION_SYNC_HEALTH_URL=http://federation-sync:8000/sync/health +# FEDERATION_SYNC_USER_EMAIL=federation-sync@internal.local +# Comma-separated CIDRs allowed to call export (default: private Docker ranges). +# FEDERATION_EXPORT_ALLOWED_CIDRS= + # AUTH0 # ------------------------------------------------------------------------------ # Set these from your Auth0 application settings diff --git a/gateway/compose.local.yaml b/gateway/compose.local.yaml index fd9343a5d..c9d4d25aa 100644 --- a/gateway/compose.local.yaml +++ b/gateway/compose.local.yaml @@ -83,6 +83,7 @@ services: - ./.envs/local/storage.env - ./.envs/local/postgres.env - ./.envs/local/opensearch.env + - ../federation-shared.env environment: - SKIP_POSTGRES_WAIT=1 # remember /entrypoint runs first diff --git a/gateway/config/api_router.py b/gateway/config/api_router.py index cc9a80f44..b181398ea 100644 --- a/gateway/config/api_router.py +++ b/gateway/config/api_router.py @@ -5,6 +5,7 @@ from sds_gateway.api_methods.views.auth_endpoints import ValidateAuthViewSet from sds_gateway.api_methods.views.capture_endpoints import CaptureViewSet from sds_gateway.api_methods.views.dataset_endpoints import DatasetViewSet +from sds_gateway.api_methods.views.federation_endpoints import FederationViewSet from sds_gateway.api_methods.views.file_endpoints import FileViewSet from sds_gateway.api_methods.views.file_endpoints import check_contents_exist from sds_gateway.users.api.views import UserViewSet @@ -17,6 +18,7 @@ router.register(r"assets/files", FileViewSet, basename="files") router.register(r"assets/captures", CaptureViewSet, basename="captures") router.register(r"assets/datasets", DatasetViewSet, basename="datasets") +router.register(r"federation", FederationViewSet, basename="federation") if settings.VISUALIZATIONS_ENABLED: router.register(r"visualizations", VisualizationViewSet, basename="visualizations") diff --git a/gateway/config/settings/base.py b/gateway/config/settings/base.py index 87fb85ab4..92efe5e4d 100644 --- a/gateway/config/settings/base.py +++ b/gateway/config/settings/base.py @@ -1,6 +1,7 @@ """Base settings to build other settings files upon.""" # ruff: noqa: ERA001 +import ipaddress import random import string from pathlib import Path @@ -9,6 +10,9 @@ from celery.schedules import crontab from environs import env +from sds_gateway.api_methods.federation.redis_channel import ( + resolve_federation_events_channel, +) from config.settings.logs import ColoredFormatter from config.settings.utils import guess_admin_console_env @@ -24,6 +28,41 @@ def __get_random_token(length: int) -> str: ) +FEDERATION_EXPORT_ALLOWED_CIDRS_DEFAULT: list[str] = [ + "127.0.0.1/32", + "::1/128", + "10.0.0.0/8", + "172.16.0.0/12", + "192.168.0.0/16", +] + + +def _parse_cidrs( + raw: list[str], +) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]: + networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = [] + for item in raw: + text = str(item).strip() + if not text: + continue + networks.append(ipaddress.ip_network(text, strict=False)) + return networks + + +def _federation_export_allowed_cidrs_from_env() -> list[ + ipaddress.IPv4Network | ipaddress.IPv6Network +]: + raw = env.list( + "FEDERATION_EXPORT_ALLOWED_CIDRS", + default=FEDERATION_EXPORT_ALLOWED_CIDRS_DEFAULT, + ) + tokens = [str(item).strip() for item in raw] + tokens = [token for token in tokens if token] + if not tokens: + tokens = list(FEDERATION_EXPORT_ALLOWED_CIDRS_DEFAULT) + return _parse_cidrs(tokens) + + env.read_env() SITE_DOMAIN: str = env.str("SITE_DOMAIN", default="localhost:8000") @@ -610,7 +649,10 @@ def _strip_endpoint_scheme(endpoint_url: str) -> str: "rest_framework.authentication.SessionAuthentication", "sds_gateway.api_methods.authentication.APIKeyAuthentication", ), - "DEFAULT_PERMISSION_CLASSES": ("rest_framework.permissions.IsAuthenticated",), + "DEFAULT_PERMISSION_CLASSES": ( + "rest_framework.permissions.IsAuthenticated", + "sds_gateway.api_methods.permissions.DisallowFederationSyncKey", + ), "DEFAULT_SCHEMA_CLASS": "drf_spectacular.openapi.AutoSchema", "DEFAULT_THROTTLE_RATES": { "vis_stream": VIS_STREAM_THROTTLE_RATE, @@ -709,6 +751,56 @@ def _strip_endpoint_scheme(endpoint_url: str) -> str: SDS_PROGRAMMATIC_SITE_NAME: str = env.str("SDS_PROGRAMMATIC_SITE_NAME", default="sds") SDS_SITE_FQDN: str = env.str("SDS_SITE_FQDN", default="localhost") +# Federation peer short name (RFC [site].name, e.g. crc, haystack); +# not SDS_PROGRAMMATIC_SITE_NAME. +FEDERATION_SITE_NAME: str = env.str("FEDERATION_SITE_NAME", default="").strip() +# Master switch: when False, federation export and Redis events are inactive. +FEDERATION_ENABLED: bool = env.bool("FEDERATION_ENABLED", default=False) +_federation_events_channel_override: str = env.str( + "FEDERATION_EVENTS_CHANNEL", + default="", +).strip() +FEDERATION_EVENTS_CHANNEL: str = resolve_federation_events_channel( + site_name=FEDERATION_SITE_NAME, + channel_override=_federation_events_channel_override, +) +FEDERATION_SYNC_USER_EMAIL: str = env.str( + "FEDERATION_SYNC_USER_EMAIL", + default="federation-sync@internal.local", +) +# From federation-shared.env (compose env_file); +# used by init_federation_sync_token only. +FEDERATION_SYNC_DRF_TOKEN: str = env.str("FEDERATION_SYNC_DRF_TOKEN", default="") +FEDERATION_SYNC_HEALTH_URL: str = env.str( + "FEDERATION_SYNC_HEALTH_URL", + default="http://federation-sync:8000/sync/health", +) +FEDERATION_SYNC_HEALTH_PROBE_TIMEOUT: float = env.float( + "FEDERATION_SYNC_HEALTH_PROBE_TIMEOUT", + default=2.0, +) +FEDERATION_SKIP_SYNC_HEALTH_PROBE: bool = env.bool( + "FEDERATION_SKIP_SYNC_HEALTH_PROBE", + default=False, +) +FEDERATION_SKIP_SYNC_API_KEY_CHECK: bool = env.bool( + "FEDERATION_SKIP_SYNC_API_KEY_CHECK", + default=False, +) +FEDERATION_SKIP_REDIS_PROBE: bool = env.bool( + "FEDERATION_SKIP_REDIS_PROBE", + default=False, +) +# Set at startup / periodic recheck by federation.availability. +FEDERATION_OPERATIONAL: bool = False +FEDERATION_OPERATIONAL_REASON: str = "" +# Tests may set via override_settings without running probes. +FEDERATION_OPERATIONAL_OVERRIDE: bool | None = None +# Export API: internal Docker/private networks (sync → django on sds-network). +FEDERATION_EXPORT_ALLOWED_CIDRS: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = ( + _federation_export_allowed_cidrs_from_env() +) + # ADMIN_CONSOLE_ENV is used to visually distinguish between different environments # (production, staging, local) in the admin console and error emails. It does not affect # any functionality and it is meant to prevent changes in production meant for testing diff --git a/gateway/pyproject.toml b/gateway/pyproject.toml index 4ed42b8b8..a88f1bdee 100644 --- a/gateway/pyproject.toml +++ b/gateway/pyproject.toml @@ -147,6 +147,10 @@ # https://deptry.com/usage/#per-rule-ignores [tool.deptry.per_rule_ignores] + DEP001 = [ + # optional monorepo sibling; contract tests add federation/ to sys.path + "sds_federation", + ] DEP002 = [ # packages that are installed but not imported "argon2-cffi", # used by django for argon2 password hashing diff --git a/gateway/scripts/fallow-cross-file-dupes.sh b/gateway/scripts/fallow-cross-file-dupes.sh index 5d2e91c14..c1ac45594 100755 --- a/gateway/scripts/fallow-cross-file-dupes.sh +++ b/gateway/scripts/fallow-cross-file-dupes.sh @@ -2,24 +2,31 @@ set -euo pipefail cd "$(dirname "$0")/.." -# Use first available: vpx > bunx > npx -if command -v pnpx &>/dev/null; then - RUNNER=pnpx -elif command -v vpx &>/dev/null; then - RUNNER=vpx -elif command -v bunx &>/dev/null; then - RUNNER=bunx -elif command -v npx &>/dev/null; then - RUNNER=npx +LOCAL_FALLOW="node_modules/.bin/fallow" +if [[ -x "${LOCAL_FALLOW}" ]]; then + "${LOCAL_FALLOW}" dupes --format json -q | jq -e ' + [ (.clone_groups // .dupes.clone_groups // [])[] + | select((.instances | map(.file) | unique | length) > 1) + ] | length == 0 +' >/dev/null else - echo "Error: neither vpx, bunx, nor npx found in PATH" >&2 - exit 1 -fi + # Use first available: vpx > bunx > npx (avoid pnpx; global pnpm may need newer Node) + if command -v vpx &>/dev/null; then + RUNNER=(vpx) + elif command -v bunx &>/dev/null; then + RUNNER=(bunx) + elif command -v npx &>/dev/null; then + RUNNER=(npx) + else + echo "Error: neither local fallow, vpx, bunx, nor npx found" >&2 + exit 1 + fi -"${RUNNER}" fallow dupes --format json -q | jq -e ' + "${RUNNER[@]}" fallow dupes --format json -q | jq -e ' [ (.clone_groups // .dupes.clone_groups // [])[] | select((.instances | map(.file) | unique | length) > 1) ] | length == 0 ' >/dev/null +fi echo "No cross-file clone groups detected." diff --git a/gateway/scripts/generate-secrets.sh b/gateway/scripts/generate-secrets.sh index ea6e180c7..cdb74318f 100755 --- a/gateway/scripts/generate-secrets.sh +++ b/gateway/scripts/generate-secrets.sh @@ -3,8 +3,12 @@ set -Eeuo pipefail SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) GATEWAY_ROOT=$(cd "${SCRIPT_DIR}/.." && pwd) -SFS_ROOT=$(cd "${GATEWAY_ROOT}/../seaweedfs" && pwd) +SDS_ROOT=$(cd "${GATEWAY_ROOT}/.." && pwd) +FEDERATION_ROOT=$(cd "${SDS_ROOT}/federation" && pwd) +SFS_ROOT=$(cd "${SDS_ROOT}/seaweedfs" && pwd) EXAMPLE_DIR="${GATEWAY_ROOT}/.envs/example" +FEDERATION_SHARED_TEMPLATE="${SDS_ROOT}/federation-shared.example.env" +FEDERATION_SHARED_OUTPUT="${SDS_ROOT}/federation-shared.env" # PRIMARY (RustFS or SeaweedFS) PRIMARY_ACCESS_KEY_ID="" @@ -16,6 +20,20 @@ PRIMARY_S3_ENDPOINT_URL="" SECONDARY_ACCESS_KEY_ID="" SECONDARY_SECRET_ACCESS_KEY="" +FEDERATION_SYNC_DRF_TOKEN="" + +function ensure_federation_sync_drf_token() { + local env_type="$1" + if [[ -n "${FEDERATION_SYNC_DRF_TOKEN}" ]]; then + return 0 + fi + if [[ "${env_type}" == "ci" ]]; then + FEDERATION_SYNC_DRF_TOKEN="ci-federation-sync-drf-01234567890123456" + else + FEDERATION_SYNC_DRF_TOKEN=$(generate_secret 40) + fi +} + function usage() { cat < @@ -35,7 +53,8 @@ EXAMPLES: ${0} production # Generate production env files NOTES: - - Generated files are placed in .envs// directory + - Generated gateway files are placed in gateway/.envs// + - federation-shared.env is written to the repo root (from federation-shared.example.env) - Example templates are read from .envs/example/ - Secrets are randomly generated using OpenSSL - CI environment uses insecure but deterministic values for ephemeral usage @@ -120,7 +139,7 @@ function process_env_file() { return 0 fi - echo " ✓ Generating ${output}" + echo " ✓ Generating ${output} (from ${template})" local content content=$(cat "${template}") @@ -167,6 +186,11 @@ function process_env_file() { # set WEB_CONCURRENCY based on CPU cores (applies to all environments) content="${content//WEB_CONCURRENCY=4/WEB_CONCURRENCY=${web_concurrency}}" + if [[ "${filename}" == "federation-shared.example.env" ]]; then + ensure_federation_sync_drf_token "${env_type}" + content="${content//FEDERATION_SYNC_DRF_TOKEN=/FEDERATION_SYNC_DRF_TOKEN=${FEDERATION_SYNC_DRF_TOKEN}}" + fi + if [[ "${filename}" == "storage.env" ]]; then # PRIMARY vars content="${content//PRIMARY_ACCESS_KEY_ID=admin/PRIMARY_ACCESS_KEY_ID=${PRIMARY_ACCESS_KEY_ID}}" @@ -232,7 +256,11 @@ function process_env_file() { # content="${content//RUSTFS_SECRET_ACCESS_KEY=/RUSTFS_SECRET_ACCESS_KEY=${SECONDARY_SECRET_ACCESS_KEY}}" fi - # write to output + # write to output (never overwrite *.example.env templates) + if [[ "${output}" == *".example.env" ]]; then + echo "ERROR: refused to write secrets to example template: ${output}" >&2 + exit 1 + fi mkdir -p "$(dirname "${output}")" echo "${content}" >"${output}" chmod 600 "${output}" @@ -243,12 +271,16 @@ function set_permissions() { env_dirs=( "${GATEWAY_ROOT}/.envs" "${SFS_ROOT}/.envs" + "${FEDERATION_ROOT}/.envs" ) for dir in "${env_dirs[@]}"; do if [ -d "${dir}" ]; then find "${dir}" -type f -name "*.env" -exec chmod --changes 600 {} \; fi done + if [[ -f "${FEDERATION_SHARED_OUTPUT}" ]]; then + chmod --changes 600 "${FEDERATION_SHARED_OUTPUT}" + fi } function main() { @@ -326,10 +358,24 @@ function main() { process_env_file "${template}" "${output}" "${env_type}" "${force}" done + if [[ -f "${FEDERATION_SHARED_TEMPLATE}" ]]; then + # Read federation-shared.example.env → write repo-root federation-shared.env only + process_env_file \ + "${FEDERATION_SHARED_TEMPLATE}" \ + "${FEDERATION_SHARED_OUTPUT}" \ + "${env_type}" \ + "${force}" + else + echo " ⏭ ${FEDERATION_SHARED_TEMPLATE} not found (skipping federation-shared.env)" + fi + set_permissions echo "" echo "✅ Secrets generated successfully in ${target_dir_gwy}/" + if [[ -f "${FEDERATION_SHARED_OUTPUT}" ]]; then + echo " Federation shared env: ${FEDERATION_SHARED_OUTPUT}" + fi echo "" echo "Next steps:" if [[ "${env_type}" == "ci" ]]; then diff --git a/gateway/sds_gateway/api_methods/apps.py b/gateway/sds_gateway/api_methods/apps.py index 73073e45a..a9b6150af 100644 --- a/gateway/sds_gateway/api_methods/apps.py +++ b/gateway/sds_gateway/api_methods/apps.py @@ -1,9 +1,24 @@ import logging +import sys from django.apps import AppConfig +from django.db.models.signals import post_migrate from loguru import logger as log +def _skip_federation_init_in_ready() -> bool: + """Management commands that load apps before the schema exists.""" + if len(sys.argv) < 2: # noqa: PLR2004 + return False + return sys.argv[1] in { + "migrate", + "makemigrations", + "showmigrations", + "sqlmigrate", + "flush", + } + + class ApiMethodsConfig(AppConfig): default_auto_field = "django.db.models.BigAutoField" name = "sds_gateway.api_methods" @@ -12,7 +27,23 @@ class ApiMethodsConfig(AppConfig): # pattern to import application modules here in ready() # ruff: noqa: PLC0415 def ready(self) -> None: + import sds_gateway.api_methods.federation.signals import sds_gateway.api_methods.schema # noqa: F401 + from sds_gateway.api_methods.federation.availability import ( + initialize_federation_operational_state, + ) + + def _init_federation_after_migrate(sender, **kwargs) -> None: + initialize_federation_operational_state() + + post_migrate.connect( + _init_federation_after_migrate, + sender=self, + dispatch_uid="api_methods_federation_operational_init", + ) + + if not _skip_federation_init_in_ready(): + initialize_federation_operational_state() silence_unwanted_logs() diff --git a/gateway/sds_gateway/api_methods/authentication.py b/gateway/sds_gateway/api_methods/authentication.py index 2a210478f..7dfa520a9 100644 --- a/gateway/sds_gateway/api_methods/authentication.py +++ b/gateway/sds_gateway/api_methods/authentication.py @@ -13,7 +13,7 @@ class APIKeyAuthentication(BaseAuthentication): keyword = "Api-Key" - def authenticate(self, request) -> tuple[User, bool]: + def authenticate(self, request) -> tuple[User, UserAPIKey]: """Authenticates the user with their API key. Args: request: Contains the API key in the Authorization header. @@ -25,8 +25,7 @@ def authenticate(self, request) -> tuple[User, bool]: """ auth_header = request.headers.get("Authorization") if not auth_header: - msg = "No API key provided" - raise AuthenticationFailed(msg) + return None # extract key from header split_header = auth_header.split(":") @@ -47,7 +46,7 @@ def authenticate(self, request) -> tuple[User, bool]: raise AuthenticationFailed(msg) from err user = api_key_obj.user - return (user, True) + return (user, api_key_obj) def authenticate_header(self, request) -> str: return self.keyword diff --git a/gateway/sds_gateway/api_methods/federation/__init__.py b/gateway/sds_gateway/api_methods/federation/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/gateway/sds_gateway/api_methods/federation/availability.py b/gateway/sds_gateway/api_methods/federation/availability.py new file mode 100644 index 000000000..dcca8aada --- /dev/null +++ b/gateway/sds_gateway/api_methods/federation/availability.py @@ -0,0 +1,203 @@ +"""Federation operational status: config, sync health, Redis, sync API key.""" + +from __future__ import annotations + +import ipaddress +import json +import time +import urllib.error +import urllib.request +from typing import Any + +from config.settings.base import FEDERATION_EXPORT_ALLOWED_CIDRS_DEFAULT +from config.settings.base import _parse_cidrs +from django.conf import settings +from django.db import connection +from django.db.utils import DatabaseError +from loguru import logger as log + +from sds_gateway.api_methods.models import KeySources +from sds_gateway.api_methods.tasks import get_redis_client +from sds_gateway.users.models import UserAPIKey + +_HTTP_OK = 200 +_RECHECK_INTERVAL_SECONDS = 60.0 +_last_evaluated_at: float = 0.0 +_cached_operational: bool = False +_cached_reason: str = "not evaluated" + + +def _setting(name: str, *, default: Any = None) -> Any: + return getattr(settings, name, default) + + +def _export_allowed_networks() -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]: + """Networks from settings (parsed at startup); tests may override with strings.""" + raw = _setting("FEDERATION_EXPORT_ALLOWED_CIDRS", default=[]) + networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = [] + for item in raw: + if isinstance(item, (ipaddress.IPv4Network, ipaddress.IPv6Network)): + networks.append(item) + continue + text = str(item).strip() + if not text: + continue + networks.append(ipaddress.ip_network(text, strict=False)) + if not networks: + networks = _parse_cidrs(FEDERATION_EXPORT_ALLOWED_CIDRS_DEFAULT) + return networks + + +def federation_client_ip(request) -> str | None: + """Client IP for export access control (direct internal connections only).""" + remote = request.META.get("REMOTE_ADDR") + if remote: + return str(remote).strip() + return None + + +def is_client_ip_allowed_for_federation_export(request) -> bool: + cidrs = _export_allowed_networks() + if not cidrs: + return False + client_ip = federation_client_ip(request) + if not client_ip: + return False + try: + addr = ipaddress.ip_address(client_ip) + except ValueError: + return False + return any(addr in network for network in cidrs) + + +def _sync_health_ok() -> tuple[bool, str]: # noqa: C901, PLR0911 + if _setting("FEDERATION_SKIP_SYNC_HEALTH_PROBE", default=False): + return True, "health probe skipped" + url = (_setting("FEDERATION_SYNC_HEALTH_URL") or "").strip() + if not url: + return False, "FEDERATION_SYNC_HEALTH_URL is not set" + if not url.startswith(("http://", "https://")): + return False, "FEDERATION_SYNC_HEALTH_URL must be http(s)" + timeout = float( + _setting("FEDERATION_SYNC_HEALTH_PROBE_TIMEOUT", default=2.0), + ) + request = urllib.request.Request(url, method="GET") # noqa: S310 + try: + with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 + if response.status != _HTTP_OK: + return False, f"sync health returned HTTP {response.status}" + body = response.read().decode("utf-8", errors="replace") + except urllib.error.URLError as exc: + return False, f"sync health probe failed: {exc.reason}" + except TimeoutError: + return False, "sync health probe timed out" + if not body.strip(): + return True, "sync health returned 200" + + try: + payload = json.loads(body) + except json.JSONDecodeError: + return True, "sync health returned 200" + + if isinstance(payload, dict): + if payload.get("status") == "ok": + return True, "sync health ok" + status_value = payload.get("status") + return False, f"sync health status is not ok: {status_value!r}" + + return True, "sync health returned 200" + + +def _sync_api_key_present() -> tuple[bool, str]: + if _setting("FEDERATION_SKIP_SYNC_API_KEY_CHECK", default=False): + return True, "sync API key check skipped" + exists = UserAPIKey.objects.filter( + source=KeySources.FederationSync, + revoked=False, + ).exists() + if not exists: + return False, "no FederationSync API key in database" + return True, "FederationSync API key present" + + +def _redis_ok() -> tuple[bool, str]: + if not _setting("FEDERATION_ENABLED", default=False): + return True, "redis not required (federation disabled)" + if _setting("FEDERATION_SKIP_REDIS_PROBE", default=False): + return True, "redis probe skipped" + + try: + client = get_redis_client() + client.ping() + except Exception as exc: # noqa: BLE001 + return False, f"redis ping failed: {exc}" + return True, "redis ok" + + +def evaluate_federation_operational() -> tuple[bool, str]: + if not _setting("FEDERATION_ENABLED", default=False): + return False, "FEDERATION_ENABLED is False" + + site_name = (_setting("FEDERATION_SITE_NAME", default="") or "").strip() + if not site_name: + return False, "FEDERATION_SITE_NAME must be set when federation is enabled" + + for check in (_sync_api_key_present, _redis_ok, _sync_health_ok): + ok, reason = check() + if not ok: + return False, reason + return True, "federation operational" + + +def refresh_federation_operational_state(*, force: bool = False) -> tuple[bool, str]: + global _cached_operational, _cached_reason, _last_evaluated_at # noqa: PLW0603 + + now = time.monotonic() + if ( + not force + and _last_evaluated_at + and (now - _last_evaluated_at) < _RECHECK_INTERVAL_SECONDS + ): + return _cached_operational, _cached_reason + + operational, reason = evaluate_federation_operational() + _cached_operational = operational + _cached_reason = reason + _last_evaluated_at = now + settings.FEDERATION_OPERATIONAL = operational + settings.FEDERATION_OPERATIONAL_REASON = reason + return operational, reason + + +def federation_operational_db_ready() -> bool: + """False when federation probes would hit tables that are not migrated yet.""" + if not _setting("FEDERATION_ENABLED", default=False): + return True + table = UserAPIKey._meta.db_table # noqa: SLF001 + try: + return table in connection.introspection.table_names() + except DatabaseError: + return False + + +def initialize_federation_operational_state() -> None: + if not federation_operational_db_ready(): + settings.FEDERATION_OPERATIONAL = False + settings.FEDERATION_OPERATIONAL_REASON = "database tables not ready" + log.debug("Federation operational init deferred until migrations apply") + return + + operational, reason = refresh_federation_operational_state(force=True) + if operational: + log.info("Federation is operational: {}", reason) + else: + log.warning("Federation disabled: {}", reason) + + +def is_federation_operational() -> bool: + if _setting("FEDERATION_OPERATIONAL_OVERRIDE", default=None) is not None: + return bool(_setting("FEDERATION_OPERATIONAL_OVERRIDE")) + if not _setting("FEDERATION_ENABLED", default=False): + return False + operational, _reason = refresh_federation_operational_state() + return operational diff --git a/gateway/sds_gateway/api_methods/federation/compile_federated_data.py b/gateway/sds_gateway/api_methods/federation/compile_federated_data.py new file mode 100644 index 000000000..29be27ffb --- /dev/null +++ b/gateway/sds_gateway/api_methods/federation/compile_federated_data.py @@ -0,0 +1,140 @@ +"""Build federation export payloads and fed-* OpenSearch helpers.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from typing import Any + +from django.conf import settings +from loguru import logger as log +from opensearchpy import exceptions as os_exceptions +from opensearchpy.exceptions import NotFoundError + +from sds_gateway.api_methods.federation.fed_index import federated_doc_id +from sds_gateway.api_methods.federation.fed_index import index_for_item_type +from sds_gateway.api_methods.models import Capture +from sds_gateway.api_methods.models import Dataset +from sds_gateway.api_methods.models import ItemType +from sds_gateway.api_methods.serializers.capture_serializers import ( + CaptureFederationSerializer, +) +from sds_gateway.api_methods.serializers.dataset_serializers import ( + DatasetFederationSerializer, +) +from sds_gateway.api_methods.utils.opensearch_client import get_opensearch_client + +if TYPE_CHECKING: + from uuid import UUID + + from django.db.models import QuerySet + + +def federation_site_name() -> str: + return getattr(settings, "FEDERATION_SITE_NAME", "").strip() + + +def capture_in_published_dataset(capture: Capture) -> bool: + return capture.datasets.federation_exportable().exists() + + +def capture_in_other_published_datasets( + capture: Capture, + *, + exclude_dataset_id: UUID, +) -> bool: + return ( + capture.datasets.federation_exportable() + .exclude(uuid=exclude_dataset_id) + .exists() + ) + + +def compile_federated_dataset_doc(dataset: Dataset) -> dict[str, Any]: + site_name = federation_site_name() + return DatasetFederationSerializer( + dataset, + context={"site_name": site_name}, + ).data + + +def compile_federated_capture_doc(capture: Capture) -> dict[str, Any]: + site_name = federation_site_name() + return CaptureFederationSerializer( + capture, + context={"site_name": site_name}, + ).data + + +def compile_federated_doc( + instance: Dataset | Capture, +) -> dict[str, Any]: + if isinstance(instance, Dataset): + return compile_federated_dataset_doc(instance) + return compile_federated_capture_doc(instance) + + +def get_federated_export_doc_by_uuid( + *, + asset_type: ItemType, + asset_uuid: UUID, + site_name: str | None = None, +) -> dict[str, Any] | None: + """Return indexed federation export body for this site, or None if missing.""" + resolved_site = (site_name or federation_site_name()).strip() + if not resolved_site: + return None + client = get_opensearch_client() + index = index_for_item_type(asset_type) + doc_id = federated_doc_id( + site_name=resolved_site, + uuid=asset_uuid, + ) + try: + response = client.get(index=index, id=doc_id) + except NotFoundError: + return None + except os_exceptions.OpenSearchException as exc: + log.warning( + "Federation fed doc lookup failed for {} {}: {}", + asset_type.value, + asset_uuid, + exc, + ) + return None + source = response.get("_source") + if not isinstance(source, dict): + return None + return source + + +def fed_doc_exists( + *, + asset_type: ItemType, + site_name: str, + instance: Dataset | Capture, +) -> bool: + return ( + get_federated_export_doc_by_uuid( + asset_type=asset_type, + asset_uuid=instance.uuid, + site_name=site_name, + ) + is not None + ) + + +def public_datasets_queryset() -> QuerySet[Dataset]: + return ( + Dataset.objects.federation_exportable() + .prefetch_related("keywords", "owner") + .order_by("-updated_at") + ) + + +def public_captures_queryset() -> QuerySet[Capture]: + return ( + Capture.objects.filter(is_deleted=False) + .filter(datasets__in=Dataset.objects.federation_exportable()) + .distinct() + .order_by("-updated_at") + ) diff --git a/gateway/sds_gateway/api_methods/federation/events.py b/gateway/sds_gateway/api_methods/federation/events.py new file mode 100644 index 000000000..2e7d5760d --- /dev/null +++ b/gateway/sds_gateway/api_methods/federation/events.py @@ -0,0 +1,43 @@ +"""Publish federation change notifications to Redis.""" + +from __future__ import annotations + +import json +from datetime import UTC +from datetime import datetime +from typing import TYPE_CHECKING +from typing import Any + +from django.conf import settings +from loguru import logger as log + +from sds_gateway.api_methods.federation.availability import is_federation_operational +from sds_gateway.api_methods.tasks import get_redis_client + +if TYPE_CHECKING: + from uuid import UUID + + from sds_gateway.api_methods.models import ItemType + + +def publish_federation_event( + *, + item_type: ItemType, + uuid: UUID, + timestamp: datetime | None = None, +) -> None: + """Notify the local federation sync service via Redis pub/sub.""" + if not is_federation_operational(): + log.debug("Federation not operational, skipping Redis publish") + return + channel = settings.FEDERATION_EVENTS_CHANNEL + payload: dict[str, Any] = { + "item_type": item_type.value, + "uuid": str(uuid), + "timestamp": (timestamp or datetime.now(UTC)).isoformat(), + } + try: + client = get_redis_client() + client.publish(channel, json.dumps(payload)) + except Exception as err: # noqa: BLE001 + log.warning("Failed to publish federation event: {}", err) diff --git a/gateway/sds_gateway/api_methods/federation/export_contract.py b/gateway/sds_gateway/api_methods/federation/export_contract.py new file mode 100644 index 000000000..0a7d65df1 --- /dev/null +++ b/gateway/sds_gateway/api_methods/federation/export_contract.py @@ -0,0 +1,32 @@ +"""Helpers to keep gateway export serializers aligned with sync Pydantic models.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from typing import Any + +if TYPE_CHECKING: + from pydantic import BaseModel + from rest_framework.serializers import BaseSerializer + + +def serializer_output_field_names(serializer: BaseSerializer[Any]) -> set[str]: + return set(serializer.fields.keys()) + + +def assert_field_names_match( + serializer: BaseSerializer[Any], + pydantic_model: BaseModel, + *, + label: str, +) -> None: + expected = set(pydantic_model.model_fields.keys()) + actual = serializer_output_field_names(serializer) + if expected != actual: + missing = expected - actual + extra = actual - expected + msg = ( + f"{label} field mismatch: missing={sorted(missing)!r} " + f"extra={sorted(extra)!r}" + ) + raise AssertionError(msg) diff --git a/gateway/sds_gateway/api_methods/federation/fed_index.py b/gateway/sds_gateway/api_methods/federation/fed_index.py new file mode 100644 index 000000000..3a338a4da --- /dev/null +++ b/gateway/sds_gateway/api_methods/federation/fed_index.py @@ -0,0 +1,59 @@ +"""Write local site documents into fed-* OpenSearch indices.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from typing import Any + +from sds_gateway.api_methods.models import ItemType + +if TYPE_CHECKING: + from datetime import datetime + from uuid import UUID + + from opensearchpy import OpenSearch + +FED_DATASETS_INDEX = "fed-datasets" +FED_CAPTURES_INDEX = "fed-captures" + + +def federated_doc_id(site_name: str, uuid: UUID) -> str: + return f"{site_name}:{uuid}" + + +def index_for_item_type(item_type: ItemType) -> str: + if item_type == ItemType.DATASET: + return FED_DATASETS_INDEX + if item_type == ItemType.CAPTURE: + return FED_CAPTURES_INDEX + msg = f"unsupported federation item type: {item_type}" + raise ValueError(msg) + + +class LocalFederatedIndexer: + """Index this site's public federation export shape into shared fed-* indices.""" + + def __init__(self, client: OpenSearch) -> None: + self._client = client + + def apply_local_event( + self, + *, + event_at: datetime, + site_name: str, + item_type: ItemType, + uuid: UUID, + body: dict[str, Any], + ) -> None: + index_name = index_for_item_type(item_type) + doc_id = federated_doc_id(site_name, uuid) + doc = { + **body, + "federation_event_at": event_at.isoformat(), + } + self._client.index( + index=index_name, + id=doc_id, + body=doc, + refresh="wait_for", + ) diff --git a/gateway/sds_gateway/api_methods/federation/permissions.py b/gateway/sds_gateway/api_methods/federation/permissions.py new file mode 100644 index 000000000..2936c2e62 --- /dev/null +++ b/gateway/sds_gateway/api_methods/federation/permissions.py @@ -0,0 +1,39 @@ +"""Permissions specific to federation export endpoints.""" + +from django.conf import settings +from rest_framework.exceptions import APIException +from rest_framework.permissions import BasePermission + +from sds_gateway.api_methods.federation.availability import ( + is_client_ip_allowed_for_federation_export, +) +from sds_gateway.api_methods.federation.availability import is_federation_operational + + +class FederationNotOperational(APIException): + status_code = 503 + default_detail = "Federation is not configured or the sync service is unavailable." + default_code = "federation_unavailable" + + +class IsFederationOperational(BasePermission): + """Deny export when federation failed startup / health checks.""" + + message = FederationNotOperational.default_detail + + def has_permission(self, request, view) -> bool: + if not is_federation_operational(): + detail = getattr(settings, "FEDERATION_OPERATIONAL_REASON", "") or ( + FederationNotOperational.default_detail + ) + raise FederationNotOperational(detail=detail) + return True + + +class IsFederationInternalExportClient(BasePermission): + """Restrict export to internal network clients (sync service on sds-network).""" + + message = "Federation export is only available to internal clients." + + def has_permission(self, request, view) -> bool: + return is_client_ip_allowed_for_federation_export(request) diff --git a/gateway/sds_gateway/api_methods/federation/redis_channel.py b/gateway/sds_gateway/api_methods/federation/redis_channel.py new file mode 100644 index 000000000..48abf17c0 --- /dev/null +++ b/gateway/sds_gateway/api_methods/federation/redis_channel.py @@ -0,0 +1,22 @@ +"""Redis pub/sub channel naming for federation change events (RFC §8).""" + +from __future__ import annotations + + +def resolve_federation_events_channel( + *, + site_name: str = "", + channel_override: str = "", +) -> str: + """Return the Redis channel for local federation events. + + Override ``channel_override`` when set (``FEDERATION_EVENTS_CHANNEL`` env). + Otherwise use ``federation:events:{site_name}`` when ``site_name`` is set. + """ + override = (channel_override or "").strip() + if override: + return override + site = (site_name or "").strip() + if site: + return f"federation:events:{site}" + return "" diff --git a/gateway/sds_gateway/api_methods/federation/reindex.py b/gateway/sds_gateway/api_methods/federation/reindex.py new file mode 100644 index 000000000..66c14360b --- /dev/null +++ b/gateway/sds_gateway/api_methods/federation/reindex.py @@ -0,0 +1,182 @@ +"""Reindex federation export documents and notify sync.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from django.db import transaction +from loguru import logger as log + +from sds_gateway.api_methods.federation.availability import is_federation_operational +from sds_gateway.api_methods.federation.compile_federated_data import ( + capture_in_published_dataset, +) +from sds_gateway.api_methods.federation.compile_federated_data import ( + compile_federated_doc, +) +from sds_gateway.api_methods.federation.compile_federated_data import fed_doc_exists +from sds_gateway.api_methods.federation.compile_federated_data import ( + federation_site_name, +) +from sds_gateway.api_methods.federation.compile_federated_data import ( + get_federated_export_doc_by_uuid, +) +from sds_gateway.api_methods.federation.events import publish_federation_event +from sds_gateway.api_methods.federation.fed_index import LocalFederatedIndexer +from sds_gateway.api_methods.models import Capture +from sds_gateway.api_methods.models import Dataset +from sds_gateway.api_methods.models import ItemType +from sds_gateway.api_methods.utils.opensearch_client import get_opensearch_client + +if TYPE_CHECKING: + from collections.abc import Iterable + + +def reindex_federated_asset( + instance: Dataset | Capture, + item_type: ItemType, +) -> None: + if not is_federation_operational(): + log.debug("Federation indexing disabled; skipping reindex") + return + + site_name = federation_site_name() + timestamp = instance.updated_at + body = compile_federated_doc(instance) + + try: + LocalFederatedIndexer(get_opensearch_client()).apply_local_event( + event_at=timestamp, + site_name=site_name, + item_type=item_type, + uuid=instance.uuid, + body=body, + ) + except Exception as exc: # noqa: BLE001 + log.warning("Federation OpenSearch index failed: {}", exc) + return + + publish_federation_event( + item_type=item_type, + uuid=instance.uuid, + timestamp=timestamp, + ) + + +def schedule_federation_dataset_reindex(dataset: Dataset) -> None: + """Run dataset (+ exportable member captures) reindex after transaction commit.""" + dataset_pk = dataset.pk + + def _on_commit() -> None: + try: + fresh = Dataset.objects.get(pk=dataset_pk) + except Dataset.DoesNotExist: + return + try: + if not dataset_needs_federation_reindex(fresh): + return + reindex_federated_asset(fresh, ItemType.DATASET) + if fresh.is_deleted or not fresh.is_federation_exportable(): + return + for capture in fresh.captures.filter(is_deleted=False): + if capture_needs_federation_reindex(capture): + reindex_federated_asset(capture, ItemType.CAPTURE) + except Exception as exc: # noqa: BLE001 + log.warning("Federation dataset reindex on commit failed: {}", exc) + + transaction.on_commit(_on_commit, robust=True) + + +def schedule_federation_capture_reindex(capture: Capture) -> None: + """Run capture reindex after transaction commit.""" + capture_pk = capture.pk + + def _on_commit() -> None: + try: + fresh = Capture.objects.get(pk=capture_pk) + except Capture.DoesNotExist: + return + try: + if not capture_needs_federation_reindex(fresh): + return + reindex_federated_asset(fresh, ItemType.CAPTURE) + except Exception as exc: # noqa: BLE001 + log.warning("Federation capture reindex on commit failed: {}", exc) + + transaction.on_commit(_on_commit, robust=True) + + +def reindex_captures_after_dataset_unlink(capture_pks: Iterable[int]) -> None: + """Reindex captures unlinked from a dataset after transaction commit.""" + pks = list(capture_pks) + if not pks: + return + + def _on_commit() -> None: + try: + for capture in Capture.objects.filter(pk__in=pks): + if capture_needs_federation_reindex(capture): + reindex_federated_asset(capture, ItemType.CAPTURE) + except Exception as exc: # noqa: BLE001 + log.warning("Federation capture unlink reindex on commit failed: {}", exc) + + transaction.on_commit(_on_commit, robust=True) + + +def _fed_doc_shows_is_deleted( + instance: Dataset | Capture, + item_type: ItemType, +) -> bool: + site_name = federation_site_name() + if not site_name: + return False + source = get_federated_export_doc_by_uuid( + asset_type=item_type, + asset_uuid=instance.uuid, + site_name=site_name, + ) + if source is None: + return False + return source.get("is_deleted") is True + + +def dataset_needs_federation_reindex(dataset: Dataset) -> bool: + if dataset.is_deleted: + return not _fed_doc_shows_is_deleted( + dataset, ItemType.DATASET + ) and fed_doc_exists( + asset_type=ItemType.DATASET, + site_name=federation_site_name(), + instance=dataset, + ) + if dataset.is_federation_exportable(): + return True + site_name = federation_site_name() + if not site_name: + return False + return fed_doc_exists( + asset_type=ItemType.DATASET, + site_name=site_name, + instance=dataset, + ) + + +def capture_needs_federation_reindex(capture: Capture) -> bool: + if capture.is_deleted: + return not _fed_doc_shows_is_deleted( + capture, ItemType.CAPTURE + ) and fed_doc_exists( + asset_type=ItemType.CAPTURE, + site_name=federation_site_name(), + instance=capture, + ) + if capture_in_published_dataset(capture): + return True + site_name = federation_site_name() + if not site_name: + return False + return fed_doc_exists( + asset_type=ItemType.CAPTURE, + site_name=site_name, + instance=capture, + ) diff --git a/gateway/sds_gateway/api_methods/federation/signals.py b/gateway/sds_gateway/api_methods/federation/signals.py new file mode 100644 index 000000000..776d8adc8 --- /dev/null +++ b/gateway/sds_gateway/api_methods/federation/signals.py @@ -0,0 +1,39 @@ +"""Index local fed-* docs on save, then notify sync via Redis.""" + +from __future__ import annotations + +from django.db.models.signals import post_save +from django.dispatch import receiver + +from sds_gateway.api_methods.federation.availability import is_federation_operational +from sds_gateway.api_methods.federation.reindex import ( + schedule_federation_capture_reindex, +) +from sds_gateway.api_methods.federation.reindex import ( + schedule_federation_dataset_reindex, +) +from sds_gateway.api_methods.models import Capture +from sds_gateway.api_methods.models import Dataset + + +@receiver(post_save, sender=Dataset) +def federation_dataset_changed( + sender: type[Dataset], + instance: Dataset, + created: bool, # noqa: FBT001 + **kwargs, +) -> None: + if not is_federation_operational(): + return + schedule_federation_dataset_reindex(instance) + + +@receiver(post_save, sender=Capture) +def federation_capture_changed( + sender: type[Capture], + instance: Capture, + **kwargs, +) -> None: + if not is_federation_operational(): + return + schedule_federation_capture_reindex(instance) diff --git a/gateway/sds_gateway/api_methods/helpers/extract_drf_metadata.py b/gateway/sds_gateway/api_methods/helpers/extract_drf_metadata.py index 39d08d862..de6fcaa78 100644 --- a/gateway/sds_gateway/api_methods/helpers/extract_drf_metadata.py +++ b/gateway/sds_gateway/api_methods/helpers/extract_drf_metadata.py @@ -160,7 +160,7 @@ def validate_metadata_by_channel( T = typing.TypeVar("T", int, bool) -def convert_or_warn( +def convert_or_warn[T: (int, bool)]( *, value: typing.Any, target_type: type[T], diff --git a/gateway/sds_gateway/api_methods/management/commands/prepare_gateway.py b/gateway/sds_gateway/api_methods/management/commands/prepare_gateway.py index 5f96e7189..c1c46bfe9 100644 --- a/gateway/sds_gateway/api_methods/management/commands/prepare_gateway.py +++ b/gateway/sds_gateway/api_methods/management/commands/prepare_gateway.py @@ -1,5 +1,6 @@ """Run gateway startup preparation in a single Django process.""" +from django.conf import settings from django.core.management import call_command from django.core.management.base import BaseCommand @@ -13,7 +14,7 @@ class Command(BaseCommand): help = ( "Prepare the gateway for serving: migrate, init search indices, " - "SVI token, and visualization pipelines." + "service tokens, and visualization pipelines." ) def add_arguments(self, parser): @@ -41,12 +42,22 @@ def handle(self, *args, **options) -> None: ("Applying database migrations...", "migrate", {"no_input": True}), ("Initializing OpenSearch indices...", "init_indices", {}), ("Initializing SVI server token...", "init_svi_token", {}), + ] + if getattr(settings, "FEDERATION_ENABLED", False): + steps.append( + ( + "Initializing federation sync server token...", + "init_federation_sync_token", + {}, + ), + ) + steps.append( ( "Setting up django-cog pipelines...", "setup_pipelines", {"strategy": pipeline_strategy}, ), - ] + ) for label, command_name, kwargs in steps: self.stdout.write(label) diff --git a/gateway/sds_gateway/api_methods/models.py b/gateway/sds_gateway/api_methods/models.py index 8c56b3c87..280712b56 100644 --- a/gateway/sds_gateway/api_methods/models.py +++ b/gateway/sds_gateway/api_methods/models.py @@ -12,6 +12,7 @@ from blake3 import blake3 as Blake3 # noqa: N812 from django.conf import settings +from django.core.exceptions import ValidationError from django.core.signals import request_started from django.db import models from django.db.models import Count @@ -88,6 +89,7 @@ class KeySources(StrEnum): SDSWebUI = "sds_web_ui" SVIBackend = "svi_backend" SVIWebUI = "svi_web_ui" + FederationSync = "federation_sync" class ItemType(StrEnum): @@ -97,6 +99,10 @@ class ItemType(StrEnum): CAPTURE = "capture" FILE = "file" + def pluralize(self) -> str: + """Get the plural form of the item type.""" + return self.value + "s" + class ProcessingType(StrEnum): """The type of post-processing.""" @@ -927,6 +933,18 @@ def __str__(self) -> str: return self.name +class DatasetQuerySet(QuerySet["Dataset"]): + def federation_exportable(self) -> QuerySet: + return self.filter( + status=DatasetStatus.FINAL, + is_public=True, + is_deleted=False, + ) + + +DatasetManager = models.Manager.from_queryset(DatasetQuerySet) + + class Dataset(BaseModel): """ Model for datasets defined and created through the API. @@ -981,6 +999,8 @@ class Dataset(BaseModel): related_name="shared_datasets", ) + objects = DatasetManager() + def __str__(self) -> str: return self.name @@ -1018,8 +1038,57 @@ def save(self, *args, **kwargs) -> None: ) raise ValueError(msg) + self.full_clean() super().save(*args, **kwargs) + def clean(self) -> None: + super().clean() + snapshot = self._published_snapshot_from_db() + if snapshot is None: + return + + old_status, old_is_public = snapshot + if not self._was_final_status(old_status) and not self._was_public( + is_public=old_is_public, + ): + return + + if self._was_final_status(status=old_status) and self.status != old_status: + msg = "Final datasets cannot be reverted to draft." + raise ValidationError(msg) + + if ( + self._was_public(is_public=old_is_public) + and self.is_public != old_is_public + ): + msg = "Public datasets cannot be reverted to private." + raise ValidationError(msg) + + def is_federation_exportable(self) -> bool: + return ( + not self.is_deleted + and self.is_public + and self.status == DatasetStatus.FINAL + ) + + def _published_snapshot_from_db(self) -> tuple[str, bool] | None: + if not self.pk: + return None + return ( + type(self) + .objects.filter(pk=self.pk) + .values_list("status", "is_public") + .first() + ) + + @staticmethod + def _was_final_status(status: DatasetStatus) -> bool: + return status == DatasetStatus.FINAL + + @staticmethod + def _was_public(*, is_public: bool) -> bool: + return is_public + def soft_delete(self) -> None: """Soft delete this record after checking for blockers.""" from sds_gateway.api_methods.utils.asset_access_control import ( # noqa: PLC0415 diff --git a/gateway/sds_gateway/api_methods/permissions.py b/gateway/sds_gateway/api_methods/permissions.py new file mode 100644 index 000000000..2fe7bf2eb --- /dev/null +++ b/gateway/sds_gateway/api_methods/permissions.py @@ -0,0 +1,31 @@ +"""DRF permissions for API key scoping.""" + +from rest_framework.permissions import BasePermission + +from sds_gateway.api_methods.models import KeySources +from sds_gateway.users.models import UserAPIKey + + +class IsFederationSyncKey(BasePermission): + """Allow only federation sync service API keys.""" + + message = "Federation sync API key required." + + def has_permission(self, request, view) -> bool: + key = request.auth + return isinstance(key, UserAPIKey) and key.source == KeySources.FederationSync + + +class DisallowFederationSyncKey(BasePermission): + """Block federation sync keys from non-federation API routes. + + Applied globally via REST_FRAMEWORK DEFAULT_PERMISSION_CLASSES. + """ + + message = "This API key is restricted to federation export endpoints." + + def has_permission(self, request, view) -> bool: + key = request.auth + return not ( + isinstance(key, UserAPIKey) and key.source == KeySources.FederationSync + ) diff --git a/gateway/sds_gateway/api_methods/serializers/capture_serializers.py b/gateway/sds_gateway/api_methods/serializers/capture_serializers.py index fd5c4cb4d..209763f0c 100644 --- a/gateway/sds_gateway/api_methods/serializers/capture_serializers.py +++ b/gateway/sds_gateway/api_methods/serializers/capture_serializers.py @@ -883,3 +883,61 @@ def serialize_capture_or_composite( # Serialize as single capture serializer = CaptureGetSerializer(capture_data["capture"], context=context) return serializer.data + + +class CaptureFederationSerializer(serializers.ModelSerializer[Capture]): + """Public-safe capture payload for federation export (sync / OpenSearch).""" + + site_name = serializers.SerializerMethodField() + file_count = serializers.SerializerMethodField() + size = serializers.SerializerMethodField() + capture_props = serializers.SerializerMethodField() + public_dataset_ids = serializers.SerializerMethodField() + created_at = serializers.DateTimeField( + format="%Y-%m-%d %H:%M:%S%z", + read_only=True, + ) + updated_at = serializers.DateTimeField( + format="%Y-%m-%d %H:%M:%S%z", + read_only=True, + ) + deleted_at = serializers.DateTimeField( + format="%Y-%m-%d %H:%M:%S%z", + read_only=True, + ) + + class Meta: + model = Capture + fields = [ + "uuid", + "name", + "capture_type", + "channel", + "scan_group", + "top_level_dir", + "site_name", + "file_count", + "size", + "capture_props", + "public_dataset_ids", + "created_at", + "updated_at", + "is_deleted", + "deleted_at", + ] + + def get_site_name(self, obj: Capture) -> str: + return str((self.context or {})["site_name"]) + + def get_file_count(self, obj: Capture) -> int: + return int(obj.get_files_summary()["total_count"]) + + def get_size(self, obj: Capture) -> int: + return int(obj.get_files_summary()["total_size"]) + + def get_capture_props(self, obj: Capture) -> dict[str, Any]: + return obj.get_opensearch_metadata() or {} + + def get_public_dataset_ids(self, obj: Capture) -> list[str]: + qs = obj.datasets.federation_exportable() + return [str(dataset.uuid) for dataset in qs] diff --git a/gateway/sds_gateway/api_methods/serializers/dataset_serializers.py b/gateway/sds_gateway/api_methods/serializers/dataset_serializers.py index 9f57ceffa..a8deb619c 100644 --- a/gateway/sds_gateway/api_methods/serializers/dataset_serializers.py +++ b/gateway/sds_gateway/api_methods/serializers/dataset_serializers.py @@ -334,6 +334,48 @@ class Meta: ] +class DatasetFederationSerializer(DatasetPublicSerializer): + """Serializer for dataset data for federation export.""" + + site_name = serializers.SerializerMethodField() + updated_at = serializers.DateTimeField( + format=READABLE_ISO_DATE_TIME, + read_only=True, + ) + size = serializers.SerializerMethodField() + capture_count = serializers.SerializerMethodField() + capture_file_count = serializers.SerializerMethodField() + artifact_file_count = serializers.SerializerMethodField() + + class Meta(DatasetPublicSerializer.Meta): + fields = [ + *DatasetPublicSerializer.Meta.fields, + "updated_at", + "site_name", + "size", + "capture_count", + "capture_file_count", + "artifact_file_count", + "is_deleted", + "deleted_at", + ] + + def get_site_name(self, obj: Dataset) -> str: + return str((self.context or {})["site_name"]) + + def get_size(self, obj: Dataset) -> int: + return int(obj.get_dataset_file_statistics()["total_size"]) + + def get_capture_count(self, obj: Dataset) -> int: + return obj.captures.filter(is_deleted=False).count() + + def get_capture_file_count(self, obj: Dataset) -> int: + return int(obj.get_dataset_file_statistics()["captures"]) + + def get_artifact_file_count(self, obj: Dataset) -> int: + return int(obj.get_dataset_file_statistics()["artifacts"]) + + def get_dataset_serializer( dataset: Dataset, *, diff --git a/gateway/sds_gateway/api_methods/tests/factories.py b/gateway/sds_gateway/api_methods/tests/factories.py index 31cab86d2..39defd8f1 100644 --- a/gateway/sds_gateway/api_methods/tests/factories.py +++ b/gateway/sds_gateway/api_methods/tests/factories.py @@ -17,6 +17,7 @@ from factory import LazyAttribute from factory import LazyFunction from factory import Sequence +from factory import SubFactory from factory import post_generation from factory.django import DjangoModelFactory from faker import Faker as FakerInstance @@ -103,7 +104,7 @@ class DatasetFactory(DjangoModelFactory): provenance = {"source": "test"} citation = {"title": "Test Dataset"} other = {"notes": "Test dataset"} - owner = FactoryFaker("subfactory", factory=UserFactory) + owner = SubFactory(UserFactory) is_deleted = False is_public = False @@ -197,7 +198,7 @@ class FileFactory(DjangoModelFactory): permissions = "rw-r--r--" size = FactoryFaker("random_int", min=1000, max=1000000) sum_blake3 = FactoryFaker("sha256") - owner = FactoryFaker("subfactory", factory=UserFactory) + owner = SubFactory(UserFactory) is_deleted = False @post_generation @@ -239,7 +240,7 @@ class Meta: channel = FactoryFaker("word") capture_type = "drf" top_level_dir = LazyFunction(lambda: _faker.file_path(depth=2).replace("/", "_")) - owner = FactoryFaker("subfactory", factory=UserFactory) + owner = SubFactory(UserFactory) name = FactoryFaker("slug") index_name = "captures-drf" @@ -260,8 +261,8 @@ class DRFDataFileFactory(DjangoModelFactory): permissions = "rw-r----" size = FactoryFaker("random_int", min=1000, max=1000000) sum_blake3 = FactoryFaker("sha256") - owner = FactoryFaker("subfactory", factory=UserFactory) - capture = FactoryFaker("subfactory", factory=CaptureFactory) + owner = SubFactory(UserFactory) + capture = SubFactory(CaptureFactory) is_deleted = False @post_generation @@ -310,8 +311,8 @@ class UserSharePermissionFactory(DjangoModelFactory): permission = UserSharePermissionFactory(is_enabled=False) """ - owner = FactoryFaker("subfactory", factory=UserFactory) - shared_with = FactoryFaker("subfactory", factory=UserFactory) + owner = SubFactory(UserFactory) + shared_with = SubFactory(UserFactory) item_type = FactoryFaker( "random_element", elements=[ItemType.DATASET, ItemType.CAPTURE] ) diff --git a/gateway/sds_gateway/api_methods/tests/test_authenticate.py b/gateway/sds_gateway/api_methods/tests/test_authenticate.py index f070fc251..7a0737720 100644 --- a/gateway/sds_gateway/api_methods/tests/test_authenticate.py +++ b/gateway/sds_gateway/api_methods/tests/test_authenticate.py @@ -34,15 +34,14 @@ def test_auth_valid_key(self) -> None: user, auth = self.auth.authenticate(request) # Verify that the user is authenticated - assert auth is True + assert isinstance(auth, UserAPIKey) assert user is not None assert user.email == "test@example.com" def test_auth_no_key(self) -> None: - """Makes sure that a request without an API key raises.""" + """Without a key, authentication is deferred to other classes.""" request = self.factory.get("/auth/") - with self.assertRaises(AuthenticationFailed): # noqa: PT027 - self.auth.authenticate(request) + assert self.auth.authenticate(request) is None def test_auth_invalid_key(self) -> None: """Makes sure that a request with an invalid API key raises.""" diff --git a/gateway/sds_gateway/api_methods/tests/test_dataset_endpoints.py b/gateway/sds_gateway/api_methods/tests/test_dataset_endpoints.py index f618ff123..6e247078a 100644 --- a/gateway/sds_gateway/api_methods/tests/test_dataset_endpoints.py +++ b/gateway/sds_gateway/api_methods/tests/test_dataset_endpoints.py @@ -638,6 +638,8 @@ def test_get_dataset_files_via_api_key(self): def test_get_dataset_files_via_session(self): """Dataset files manifest accepts session authentication.""" + self.user.is_approved = True + self.user.save(update_fields=["is_approved"]) self._attach_artifact_to_dataset() client = APIClient() client.force_login(self.user) diff --git a/gateway/sds_gateway/api_methods/tests/test_dataset_publish_immutability.py b/gateway/sds_gateway/api_methods/tests/test_dataset_publish_immutability.py new file mode 100644 index 000000000..4283851b6 --- /dev/null +++ b/gateway/sds_gateway/api_methods/tests/test_dataset_publish_immutability.py @@ -0,0 +1,45 @@ +"""Tests for Dataset publish immutability (status final / is_public).""" + +from __future__ import annotations + +import pytest +from django.core.exceptions import ValidationError + +from sds_gateway.api_methods.models import DatasetStatus +from sds_gateway.api_methods.tests.factories import DatasetFactory + +pytestmark = pytest.mark.django_db + + +class TestDatasetPublishImmutability: + def test_cannot_revert_final_to_draft(self) -> None: + dataset = DatasetFactory(status=DatasetStatus.FINAL, is_public=False) + dataset.status = DatasetStatus.DRAFT + with pytest.raises(ValidationError, match="cannot be reverted to draft"): + dataset.save() + + def test_cannot_make_public_dataset_private(self) -> None: + dataset = DatasetFactory(status=DatasetStatus.DRAFT, is_public=True) + dataset.is_public = False + with pytest.raises(ValidationError, match="cannot be reverted to private"): + dataset.save() + + def test_final_and_public_can_update_other_fields(self) -> None: + dataset = DatasetFactory( + status=DatasetStatus.FINAL, + is_public=True, + name="Original", + ) + dataset.name = "Updated title" + dataset.save() + dataset.refresh_from_db() + assert dataset.name == "Updated title" + assert dataset.status == DatasetStatus.FINAL + assert dataset.is_public is True + + def test_draft_can_still_become_final(self) -> None: + dataset = DatasetFactory(status=DatasetStatus.DRAFT, is_public=False) + dataset.status = DatasetStatus.FINAL + dataset.save() + dataset.refresh_from_db() + assert dataset.status == DatasetStatus.FINAL diff --git a/gateway/sds_gateway/api_methods/tests/test_federation_events.py b/gateway/sds_gateway/api_methods/tests/test_federation_events.py new file mode 100644 index 000000000..363b1f199 --- /dev/null +++ b/gateway/sds_gateway/api_methods/tests/test_federation_events.py @@ -0,0 +1,137 @@ +"""Tests for federation Redis event channel naming and publish.""" + +from __future__ import annotations + +import json +from contextlib import contextmanager +from unittest.mock import MagicMock +from unittest.mock import patch +from uuid import uuid4 + +import pytest +from django.test import TestCase +from django.test import override_settings + +from sds_gateway.api_methods.federation.events import publish_federation_event +from sds_gateway.api_methods.federation.redis_channel import ( + resolve_federation_events_channel, +) +from sds_gateway.api_methods.federation.signals import federation_dataset_changed +from sds_gateway.api_methods.models import Dataset +from sds_gateway.api_methods.models import DatasetStatus +from sds_gateway.api_methods.models import ItemType +from sds_gateway.api_methods.tests.factories import DatasetFactory + +pytestmark = pytest.mark.django_db + + +@contextmanager +def _federation_on_commit(): + with TestCase.captureOnCommitCallbacks(execute=True): + yield + + +class TestResolveFederationEventsChannel: + def test_site_prefixed_when_site_name_set(self) -> None: + assert ( + resolve_federation_events_channel(site_name="crc") + == "federation:events:crc" + ) + + def test_override_wins_over_site_name(self) -> None: + assert ( + resolve_federation_events_channel( + site_name="crc", + channel_override="federation:events:custom", + ) + == "federation:events:custom" + ) + + def test_empty_when_no_site_and_no_override(self) -> None: + assert resolve_federation_events_channel() == "" + + +class TestPublishFederationEvent: + @override_settings( + FEDERATION_ENABLED=True, + FEDERATION_OPERATIONAL_OVERRIDE=True, + FEDERATION_EVENTS_CHANNEL="federation:events:crc", + ) + @patch("sds_gateway.api_methods.federation.events.get_redis_client") + def test_publish_uses_configured_channel( + self, + mock_get_redis: MagicMock, + ) -> None: + mock_client = MagicMock() + mock_get_redis.return_value = mock_client + item_uuid = uuid4() + + publish_federation_event( + item_type=ItemType.DATASET, + uuid=item_uuid, + ) + + mock_client.publish.assert_called_once() + channel, payload_raw = mock_client.publish.call_args[0] + assert channel == "federation:events:crc" + payload = json.loads(payload_raw) + assert payload["item_type"] == ItemType.DATASET.value + assert payload["uuid"] == str(item_uuid) + + @override_settings( + FEDERATION_ENABLED=True, + FEDERATION_OPERATIONAL_OVERRIDE=False, + FEDERATION_EVENTS_CHANNEL="federation:events:crc", + ) + @patch("sds_gateway.api_methods.federation.events.get_redis_client") + def test_skips_publish_when_not_operational( + self, + mock_get_redis: MagicMock, + ) -> None: + publish_federation_event( + item_type=ItemType.CAPTURE, + uuid=uuid4(), + ) + mock_get_redis.assert_not_called() + + +class TestFederationSignals: + @override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_OPERATIONAL_OVERRIDE=True, + FEDERATION_EVENTS_CHANNEL="federation:events:crc", + ) + @patch("sds_gateway.api_methods.federation.reindex.publish_federation_event") + @patch("sds_gateway.api_methods.federation.reindex.LocalFederatedIndexer") + @patch( + "sds_gateway.api_methods.federation.reindex.get_opensearch_client", + new=MagicMock(), + ) + def test_dataset_post_save_indexes_when_published( + self, + mock_indexer_cls: MagicMock, + mock_publish: MagicMock, + ) -> None: + mock_indexer = mock_indexer_cls.return_value + dataset = DatasetFactory( + status=DatasetStatus.FINAL, + is_public=True, + ) + with _federation_on_commit(): + federation_dataset_changed( + sender=Dataset, + instance=dataset, + created=False, + ) + + mock_indexer.apply_local_event.assert_called_once() + call = mock_indexer.apply_local_event.call_args.kwargs + assert call["site_name"] == "crc" + assert call["item_type"] == ItemType.DATASET + assert call["uuid"] == dataset.uuid + mock_publish.assert_called_once_with( + item_type=ItemType.DATASET, + uuid=dataset.uuid, + timestamp=dataset.updated_at, + ) diff --git a/gateway/sds_gateway/api_methods/tests/test_federation_export.py b/gateway/sds_gateway/api_methods/tests/test_federation_export.py new file mode 100644 index 000000000..4d0318bf1 --- /dev/null +++ b/gateway/sds_gateway/api_methods/tests/test_federation_export.py @@ -0,0 +1,143 @@ +"""Tests for federation export endpoints and API key scoping.""" + +from unittest.mock import patch + +from django.contrib.auth import get_user_model +from django.test import override_settings +from django.urls import reverse +from rest_framework import status +from rest_framework.test import APITestCase + +from sds_gateway.api_methods.federation.compile_federated_data import ( + compile_federated_dataset_doc, +) +from sds_gateway.api_methods.models import DatasetStatus +from sds_gateway.api_methods.models import KeySources +from sds_gateway.api_methods.tests.factories import CaptureFactory +from sds_gateway.api_methods.tests.factories import DatasetFactory +from sds_gateway.users.models import UserAPIKey + +User = get_user_model() + + +@override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_OPERATIONAL_OVERRIDE=True, + FEDERATION_EXPORT_ALLOWED_CIDRS=["0.0.0.0/0", "::/0"], +) +class FederationExportAPITest(APITestCase): + def setUp(self) -> None: + self.owner = User.objects.create(email="owner@example.com", is_approved=True) + self.sync_user = User.objects.create( + email="sync@internal.local", + is_approved=True, + ) + _obj, self.sync_key = UserAPIKey.objects.create_key( + name="sync", + user=self.sync_user, + source=KeySources.FederationSync, + ) + _obj, self.user_key = UserAPIKey.objects.create_key( + name="regular", + user=self.owner, + source=KeySources.SDSWebUI, + ) + self.public_dataset = DatasetFactory( + owner=self.owner, + is_public=True, + status=DatasetStatus.FINAL, + keywords=None, + ) + self.private_dataset = DatasetFactory( + owner=self.owner, + is_public=False, + status=DatasetStatus.DRAFT, + keywords=None, + ) + self.public_capture = CaptureFactory(owner=self.owner) + self.public_capture.datasets.add(self.public_dataset) + self.list_datasets_url = reverse( + "api:federation-export-datasets-list", + ) + self.detail_dataset_url = reverse( + "api:federation-export-dataset-detail", + kwargs={"pk": str(self.public_dataset.uuid)}, + ) + self.list_captures_url = reverse( + "api:federation-export-captures-list", + ) + + def _auth(self, key: str) -> dict[str, str]: + return {"HTTP_AUTHORIZATION": f"Api-Key: {key}"} + + def test_sync_key_can_list_public_datasets(self) -> None: + response = self.client.get( + self.list_datasets_url, + REMOTE_ADDR="127.0.0.1", + **self._auth(self.sync_key), + ) + assert response.status_code == status.HTTP_200_OK + uuids = {row["uuid"] for row in response.json()} + assert str(self.public_dataset.uuid) in uuids + assert str(self.private_dataset.uuid) not in uuids + + def test_sync_key_can_retrieve_public_dataset(self) -> None: + indexed = compile_federated_dataset_doc(self.public_dataset) + with patch( + "sds_gateway.api_methods.views.federation_endpoints." + "get_federated_export_doc_by_uuid", + return_value=indexed, + ): + response = self.client.get( + self.detail_dataset_url, + REMOTE_ADDR="127.0.0.1", + **self._auth(self.sync_key), + ) + assert response.status_code == status.HTTP_200_OK + assert response.json()["uuid"] == str(self.public_dataset.uuid) + assert response.json()["site_name"] == "crc" + + def test_sync_key_dataset_detail_404_when_not_indexed(self) -> None: + with patch( + "sds_gateway.api_methods.views.federation_endpoints." + "get_federated_export_doc_by_uuid", + return_value=None, + ): + response = self.client.get( + self.detail_dataset_url, + REMOTE_ADDR="127.0.0.1", + **self._auth(self.sync_key), + ) + assert response.status_code == status.HTTP_404_NOT_FOUND + + def test_regular_key_denied_on_export(self) -> None: + response = self.client.get( + self.list_datasets_url, + REMOTE_ADDR="127.0.0.1", + **self._auth(self.user_key), + ) + assert response.status_code == status.HTTP_403_FORBIDDEN + + def test_sync_key_denied_on_dataset_assets_api(self) -> None: + url = reverse( + "api:datasets-detail", + kwargs={"pk": str(self.public_dataset.uuid)}, + ) + response = self.client.get(url, **self._auth(self.sync_key)) + assert response.status_code == status.HTTP_403_FORBIDDEN + + def test_sync_key_denied_on_capture_list(self) -> None: + url = reverse("api:captures-list") + response = self.client.get(url, **self._auth(self.sync_key)) + assert response.status_code == status.HTTP_403_FORBIDDEN + + def test_export_captures_list(self) -> None: + response = self.client.get( + self.list_captures_url, + REMOTE_ADDR="127.0.0.1", + **self._auth(self.sync_key), + ) + assert response.status_code == status.HTTP_200_OK + uuids = {row["uuid"] for row in response.json()} + assert str(self.public_capture.uuid) in uuids diff --git a/gateway/sds_gateway/api_methods/tests/test_federation_export_contract.py b/gateway/sds_gateway/api_methods/tests/test_federation_export_contract.py new file mode 100644 index 000000000..33d8bde47 --- /dev/null +++ b/gateway/sds_gateway/api_methods/tests/test_federation_export_contract.py @@ -0,0 +1,94 @@ +"""Contract tests: gateway federation export JSON ↔ sync Pydantic models.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest +from django.contrib.auth import get_user_model + +from sds_gateway.api_methods.federation.compile_federated_data import ( + compile_federated_capture_doc, +) +from sds_gateway.api_methods.federation.compile_federated_data import ( + compile_federated_dataset_doc, +) +from sds_gateway.api_methods.federation.export_contract import assert_field_names_match +from sds_gateway.api_methods.models import DatasetStatus +from sds_gateway.api_methods.serializers.capture_serializers import ( + CaptureFederationSerializer, +) +from sds_gateway.api_methods.serializers.dataset_serializers import ( + DatasetFederationSerializer, +) +from sds_gateway.api_methods.tests.factories import CaptureFactory +from sds_gateway.api_methods.tests.factories import DatasetFactory + +_repo_root = Path(__file__).resolve().parents[4] +_federation_root = _repo_root / "federation" +if _federation_root.is_dir(): + sys.path.insert(0, str(_federation_root)) + +pytest.importorskip("sds_federation") + +from sds_federation.schemas.webhooks import FederatedCaptureDoc # noqa: E402 +from sds_federation.schemas.webhooks import FederatedDatasetDoc # noqa: E402 + +User = get_user_model() + + +@pytest.mark.django_db +def test_dataset_export_field_names_match_pydantic() -> None: + owner = User.objects.create(email="owner@example.com", is_approved=True) + dataset = DatasetFactory( + owner=owner, + is_public=True, + status=DatasetStatus.FINAL, + keywords=None, + ) + serializer = DatasetFederationSerializer( + dataset, + context={"site_name": "crc"}, + ) + assert_field_names_match( + serializer, + FederatedDatasetDoc, + label="DatasetFederationSerializer", + ) + + +@pytest.mark.django_db +def test_capture_export_field_names_match_pydantic() -> None: + owner = User.objects.create(email="cap-owner@example.com", is_approved=True) + capture = CaptureFactory(owner=owner, is_public=True) + serializer = CaptureFederationSerializer( + capture, + context={"site_name": "crc"}, + ) + assert_field_names_match( + serializer, + FederatedCaptureDoc, + label="CaptureFederationSerializer", + ) + + +@pytest.mark.django_db +def test_compile_federated_dataset_doc_validates_against_pydantic() -> None: + owner = User.objects.create(email="d@example.com", is_approved=True) + dataset = DatasetFactory( + owner=owner, + is_public=True, + status=DatasetStatus.FINAL, + keywords=None, + ) + payload = compile_federated_dataset_doc(dataset) + FederatedDatasetDoc.model_validate(payload) + + +@pytest.mark.django_db +def test_compile_federated_capture_doc_validates_against_pydantic() -> None: + owner = User.objects.create(email="c@example.com", is_approved=True) + capture = CaptureFactory(owner=owner, is_public=True) + payload = compile_federated_capture_doc(capture) + FederatedCaptureDoc.model_validate(payload) diff --git a/gateway/sds_gateway/api_methods/tests/test_federation_hardening.py b/gateway/sds_gateway/api_methods/tests/test_federation_hardening.py new file mode 100644 index 000000000..f512459ab --- /dev/null +++ b/gateway/sds_gateway/api_methods/tests/test_federation_hardening.py @@ -0,0 +1,285 @@ +"""Tests for federation operational checks and export access control.""" + +from __future__ import annotations + +from unittest.mock import MagicMock +from unittest.mock import patch + +import pytest +from config.settings.base import FEDERATION_EXPORT_ALLOWED_CIDRS_DEFAULT +from config.settings.base import _federation_export_allowed_cidrs_from_env +from config.settings.base import _parse_cidrs +from django.conf import settings +from django.contrib.auth import get_user_model +from django.test import RequestFactory +from django.test import override_settings +from django.urls import reverse +from opensearchpy import exceptions as os_exceptions +from rest_framework import status +from rest_framework.test import APITestCase + +from sds_gateway.api_methods.federation.availability import ( + evaluate_federation_operational, +) +from sds_gateway.api_methods.federation.availability import ( + initialize_federation_operational_state, +) +from sds_gateway.api_methods.federation.availability import ( + is_client_ip_allowed_for_federation_export, +) +from sds_gateway.api_methods.federation.availability import ( + refresh_federation_operational_state, +) +from sds_gateway.api_methods.federation.compile_federated_data import fed_doc_exists +from sds_gateway.api_methods.models import DatasetStatus +from sds_gateway.api_methods.models import ItemType +from sds_gateway.api_methods.models import KeySources +from sds_gateway.api_methods.tests.factories import DatasetFactory +from sds_gateway.users.models import UserAPIKey + +User = get_user_model() + +pytestmark = pytest.mark.django_db + + +class TestFederationAvailability: + @override_settings(FEDERATION_ENABLED=False) + def test_disabled_when_master_switch_off(self) -> None: + ok, reason = evaluate_federation_operational() + assert ok is False + assert "FEDERATION_ENABLED" in reason + + @override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_SKIP_SYNC_API_KEY_CHECK=True, + FEDERATION_SKIP_SYNC_HEALTH_PROBE=True, + FEDERATION_SKIP_REDIS_PROBE=True, + ) + def test_operational_when_probes_skipped(self) -> None: + ok, _reason = evaluate_federation_operational() + assert ok is True + + @override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_SKIP_SYNC_HEALTH_PROBE=True, + FEDERATION_SKIP_REDIS_PROBE=True, + ) + def test_fails_without_sync_api_key(self) -> None: + ok, reason = evaluate_federation_operational() + assert ok is False + assert "FederationSync" in reason + + @override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_SKIP_SYNC_API_KEY_CHECK=True, + FEDERATION_SYNC_HEALTH_URL="http://sync.test/health", + FEDERATION_SKIP_REDIS_PROBE=True, + ) + @patch("sds_gateway.api_methods.federation.availability.urllib.request.urlopen") + def test_health_probe_success(self, mock_urlopen: MagicMock) -> None: + response = MagicMock() + response.status = 200 + response.read.return_value = b'{"status":"ok"}' + response.__enter__.return_value = response + response.__exit__.return_value = None + mock_urlopen.return_value = response + + ok, _reason = evaluate_federation_operational() + assert ok is True + + @override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_SKIP_SYNC_API_KEY_CHECK=True, + FEDERATION_SYNC_HEALTH_URL="http://sync.test/health", + FEDERATION_SKIP_REDIS_PROBE=True, + ) + @patch("sds_gateway.api_methods.federation.availability.urllib.request.urlopen") + def test_health_probe_fails_when_json_status_not_ok( + self, + mock_urlopen: MagicMock, + ) -> None: + response = MagicMock() + response.status = 200 + response.read.return_value = b'{"status":"degraded"}' + response.__enter__.return_value = response + response.__exit__.return_value = None + mock_urlopen.return_value = response + + ok, reason = evaluate_federation_operational() + assert ok is False + assert "not ok" in reason + + @override_settings( + FEDERATION_EXPORT_ALLOWED_CIDRS=["10.0.0.0/8"], + ) + def test_client_ip_allowlist(self) -> None: + factory = RequestFactory() + allowed = factory.get("/", REMOTE_ADDR="10.1.2.3") + denied = factory.get("/", REMOTE_ADDR="203.0.113.8") + assert is_client_ip_allowed_for_federation_export(allowed) is True + assert is_client_ip_allowed_for_federation_export(denied) is False + + @override_settings(FEDERATION_EXPORT_ALLOWED_CIDRS=[]) + def test_empty_cidr_override_falls_back_to_private_defaults(self) -> None: + factory = RequestFactory() + loopback = factory.get("/", REMOTE_ADDR="127.0.0.1") + public = factory.get("/", REMOTE_ADDR="203.0.113.8") + assert is_client_ip_allowed_for_federation_export(loopback) is True + assert is_client_ip_allowed_for_federation_export(public) is False + + @override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="", + FEDERATION_SKIP_SYNC_API_KEY_CHECK=True, + FEDERATION_SKIP_SYNC_HEALTH_PROBE=True, + FEDERATION_SKIP_REDIS_PROBE=True, + ) + def test_fails_without_site_name(self) -> None: + ok, reason = evaluate_federation_operational() + assert ok is False + assert "FEDERATION_SITE_NAME" in reason + + @override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_SKIP_SYNC_API_KEY_CHECK=True, + FEDERATION_SKIP_SYNC_HEALTH_PROBE=True, + FEDERATION_SKIP_REDIS_PROBE=True, + ) + def test_refresh_sets_settings_flags(self) -> None: + operational, reason = refresh_federation_operational_state(force=True) + + assert operational is True + assert settings.FEDERATION_OPERATIONAL is True + assert reason + + +class TestFederationExportCidrParsing: + def test_parse_cidrs_skips_blank_tokens(self) -> None: + networks = _parse_cidrs(["10.0.0.0/8", "", " ", "192.168.0.0/16"]) + assert len(networks) == 2 # noqa: PLR2004 + + def test_coerce_empty_env_tokens_to_defaults(self) -> None: + with patch( + "config.settings.base.env.list", + return_value=["", " "], + ): + networks = _federation_export_allowed_cidrs_from_env() + assert len(networks) == len(FEDERATION_EXPORT_ALLOWED_CIDRS_DEFAULT) + + +class TestFederationOperationalInit: + @override_settings(FEDERATION_ENABLED=True) + @patch( + "sds_gateway.api_methods.federation.availability." + "federation_operational_db_ready", + new=MagicMock(return_value=False), + ) + @patch( + "sds_gateway.api_methods.federation.availability." + "refresh_federation_operational_state", + ) + def test_initialize_skips_db_when_tables_missing( + self, + mock_refresh: MagicMock, + ) -> None: + initialize_federation_operational_state() + mock_refresh.assert_not_called() + assert settings.FEDERATION_OPERATIONAL is False + assert settings.FEDERATION_OPERATIONAL_REASON == "database tables not ready" + + +@override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_OPERATIONAL_OVERRIDE=True, + FEDERATION_EXPORT_ALLOWED_CIDRS=["0.0.0.0/0", "::/0"], +) +class FederationExportAccessControlTest(APITestCase): + """Export API with operational + network permissions enabled for tests.""" + + def setUp(self) -> None: + self.sync_user = User.objects.create( + email="sync@internal.local", + is_approved=True, + ) + _obj, self.sync_key = UserAPIKey.objects.create_key( + name="sync", + user=self.sync_user, + source=KeySources.FederationSync, + ) + owner = User.objects.create(email="owner@example.com", is_approved=True) + self.public_dataset = DatasetFactory( + owner=owner, + is_public=True, + status=DatasetStatus.FINAL, + keywords=None, + ) + self.list_datasets_url = reverse("api:federation-export-datasets-list") + + def _auth(self, key: str) -> dict[str, str]: + return {"HTTP_AUTHORIZATION": f"Api-Key: {key}"} + + def test_sync_key_allowed_from_loopback(self) -> None: + response = self.client.get( + self.list_datasets_url, + REMOTE_ADDR="127.0.0.1", + **self._auth(self.sync_key), + ) + assert response.status_code == status.HTTP_200_OK + + @override_settings( + FEDERATION_EXPORT_ALLOWED_CIDRS=["10.0.0.0/8"], + FEDERATION_OPERATIONAL_OVERRIDE=True, + ) + def test_sync_key_denied_from_public_ip(self) -> None: + response = self.client.get( + self.list_datasets_url, + REMOTE_ADDR="203.0.113.1", + **self._auth(self.sync_key), + ) + assert response.status_code == status.HTTP_403_FORBIDDEN + + @override_settings( + FEDERATION_OPERATIONAL_OVERRIDE=False, + FEDERATION_OPERATIONAL_REASON="sync health probe failed", + FEDERATION_EXPORT_ALLOWED_CIDRS=["0.0.0.0/0"], + ) + def test_export_returns_503_when_not_operational(self) -> None: + response = self.client.get( + self.list_datasets_url, + REMOTE_ADDR="127.0.0.1", + **self._auth(self.sync_key), + ) + assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE + + +class TestFederationOpenSearchReads: + @patch( + "sds_gateway.api_methods.federation.compile_federated_data.get_opensearch_client" + ) + def test_fed_doc_lookup_treats_cluster_errors_as_missing( + self, + mock_get_client: MagicMock, + ) -> None: + mock_get_client.return_value.get.side_effect = ( + os_exceptions.OpenSearchException( + 503, + "connection failed", + {}, + ) + ) + dataset = DatasetFactory(status=DatasetStatus.FINAL, is_public=True) + + assert ( + fed_doc_exists( + asset_type=ItemType.DATASET, + site_name="crc", + instance=dataset, + ) + is False + ) diff --git a/gateway/sds_gateway/api_methods/tests/test_federation_signals.py b/gateway/sds_gateway/api_methods/tests/test_federation_signals.py new file mode 100644 index 000000000..f1c150367 --- /dev/null +++ b/gateway/sds_gateway/api_methods/tests/test_federation_signals.py @@ -0,0 +1,307 @@ +"""Tests for federation signal orchestration.""" + +from __future__ import annotations + +from contextlib import contextmanager +from unittest.mock import MagicMock +from unittest.mock import patch + +import pytest +from django.db import transaction +from django.test import TestCase +from django.test import override_settings + +from sds_gateway.api_methods.federation.signals import federation_capture_changed +from sds_gateway.api_methods.federation.signals import federation_dataset_changed +from sds_gateway.api_methods.models import Capture +from sds_gateway.api_methods.models import Dataset +from sds_gateway.api_methods.models import DatasetStatus +from sds_gateway.api_methods.models import ItemType +from sds_gateway.api_methods.tests.factories import CaptureFactory +from sds_gateway.api_methods.tests.factories import DatasetFactory +from sds_gateway.api_methods.utils.asset_access_control import ( + disconnect_captures_from_dataset, +) + +pytestmark = pytest.mark.django_db + +_DATASET_AND_CAPTURE = 2 + + +@contextmanager +def _federation_on_commit(): + with TestCase.captureOnCommitCallbacks(execute=True): + yield + + +@override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_OPERATIONAL_OVERRIDE=True, +) +class TestFederationDatasetSignals(TestCase): + @patch("sds_gateway.api_methods.federation.reindex.publish_federation_event") + @patch("sds_gateway.api_methods.federation.reindex.LocalFederatedIndexer") + @patch( + "sds_gateway.api_methods.federation.reindex.get_opensearch_client", + new=MagicMock(), + ) + def test_published_dataset_upserts_and_syncs_captures( + self, + mock_indexer_cls: MagicMock, + mock_publish: MagicMock, + ) -> None: + mock_indexer = mock_indexer_cls.return_value + dataset = DatasetFactory(status=DatasetStatus.FINAL, is_public=True) + capture = CaptureFactory(is_deleted=False) + capture.datasets.add(dataset) + + with _federation_on_commit(): + federation_dataset_changed( + sender=Dataset, + instance=dataset, + created=False, + ) + + assert mock_indexer.apply_local_event.call_count == _DATASET_AND_CAPTURE + item_types = { + c.kwargs["item_type"] for c in mock_indexer.apply_local_event.call_args_list + } + assert item_types == {ItemType.DATASET, ItemType.CAPTURE} + assert mock_publish.call_count == _DATASET_AND_CAPTURE + + @patch("sds_gateway.api_methods.federation.reindex.publish_federation_event") + @patch("sds_gateway.api_methods.federation.reindex.LocalFederatedIndexer") + @patch( + "sds_gateway.api_methods.federation.reindex.get_opensearch_client", + new=MagicMock(), + ) + @patch( + "sds_gateway.api_methods.federation.reindex.get_federated_export_doc_by_uuid", + new=MagicMock(return_value={"is_deleted": False}), + ) + @patch( + "sds_gateway.api_methods.federation.reindex.fed_doc_exists", + new=MagicMock(return_value=True), + ) + def test_deleted_dataset_reindexes_full_body( + self, + mock_indexer_cls: MagicMock, + mock_publish: MagicMock, + ) -> None: + mock_indexer = mock_indexer_cls.return_value + dataset = DatasetFactory( + status=DatasetStatus.FINAL, + is_public=True, + is_deleted=True, + ) + + with _federation_on_commit(): + federation_dataset_changed( + sender=Dataset, + instance=dataset, + created=False, + ) + + mock_indexer.apply_local_event.assert_called_once() + call = mock_indexer.apply_local_event.call_args.kwargs + assert call["item_type"] == ItemType.DATASET + assert call["body"]["is_deleted"] is True + mock_publish.assert_called_once_with( + item_type=ItemType.DATASET, + uuid=dataset.uuid, + timestamp=dataset.updated_at, + ) + + @patch("sds_gateway.api_methods.federation.reindex.reindex_federated_asset") + def test_draft_dataset_skips_federation( + self, + mock_reindex: MagicMock, + ) -> None: + dataset = DatasetFactory(status=DatasetStatus.DRAFT, is_public=False) + with _federation_on_commit(): + federation_dataset_changed(sender=Dataset, instance=dataset, created=True) + mock_reindex.assert_not_called() + + @patch("sds_gateway.api_methods.federation.reindex.reindex_federated_asset") + @patch( + "sds_gateway.api_methods.federation.reindex.get_federated_export_doc_by_uuid", + new=MagicMock(return_value={"is_deleted": True}), + ) + @patch( + "sds_gateway.api_methods.federation.reindex.fed_doc_exists", + new=MagicMock(return_value=True), + ) + def test_deleted_dataset_skips_when_fed_doc_already_deleted( + self, + mock_reindex: MagicMock, + ) -> None: + dataset = DatasetFactory( + status=DatasetStatus.FINAL, + is_public=True, + is_deleted=True, + ) + with _federation_on_commit(): + federation_dataset_changed(sender=Dataset, instance=dataset, created=False) + mock_reindex.assert_not_called() + + +@override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_OPERATIONAL_OVERRIDE=True, +) +class TestFederationCaptureSignals(TestCase): + @patch("sds_gateway.api_methods.federation.reindex.publish_federation_event") + @patch("sds_gateway.api_methods.federation.reindex.LocalFederatedIndexer") + @patch( + "sds_gateway.api_methods.federation.reindex.get_opensearch_client", + new=MagicMock(), + ) + def test_capture_on_published_dataset_upserts( + self, + mock_indexer_cls: MagicMock, + mock_publish: MagicMock, + ) -> None: + mock_indexer = mock_indexer_cls.return_value + dataset = DatasetFactory(status=DatasetStatus.FINAL, is_public=True) + capture = CaptureFactory(is_deleted=False) + capture.datasets.add(dataset) + + with _federation_on_commit(): + federation_capture_changed(sender=Capture, instance=capture) + + mock_indexer.apply_local_event.assert_called_once() + assert ( + mock_indexer.apply_local_event.call_args.kwargs["item_type"] + == ItemType.CAPTURE + ) + mock_publish.assert_called_once_with( + item_type=ItemType.CAPTURE, + uuid=capture.uuid, + timestamp=capture.updated_at, + ) + + @patch("sds_gateway.api_methods.federation.reindex.reindex_federated_asset") + @patch( + "sds_gateway.api_methods.federation.reindex.fed_doc_exists", + new=MagicMock(return_value=False), + ) + def test_capture_on_draft_only_skips_without_fed_doc( + self, + mock_reindex: MagicMock, + ) -> None: + dataset = DatasetFactory(status=DatasetStatus.DRAFT, is_public=False) + capture = CaptureFactory(is_deleted=False) + capture.datasets.add(dataset) + with _federation_on_commit(): + federation_capture_changed(sender=Capture, instance=capture) + mock_reindex.assert_not_called() + + @patch("sds_gateway.api_methods.federation.reindex.publish_federation_event") + @patch("sds_gateway.api_methods.federation.reindex.LocalFederatedIndexer") + @patch( + "sds_gateway.api_methods.federation.reindex.get_opensearch_client", + new=MagicMock(), + ) + @patch( + "sds_gateway.api_methods.federation.reindex.get_federated_export_doc_by_uuid", + new=MagicMock(return_value={"is_deleted": False}), + ) + @patch( + "sds_gateway.api_methods.federation.reindex.fed_doc_exists", + new=MagicMock(return_value=True), + ) + def test_deleted_capture_reindexes_with_is_deleted( + self, + mock_indexer_cls: MagicMock, + ) -> None: + mock_indexer = mock_indexer_cls.return_value + capture = CaptureFactory(is_deleted=True) + with _federation_on_commit(): + federation_capture_changed(sender=Capture, instance=capture) + mock_indexer.apply_local_event.assert_called_once() + assert mock_indexer.apply_local_event.call_args.kwargs["body"]["is_deleted"] + + +@override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_OPERATIONAL_OVERRIDE=True, +) +class TestDatasetDisconnectReindex(TestCase): + @patch("sds_gateway.api_methods.federation.reindex.publish_federation_event") + @patch("sds_gateway.api_methods.federation.reindex.LocalFederatedIndexer") + @patch( + "sds_gateway.api_methods.federation.reindex.get_opensearch_client", + new=MagicMock(), + ) + @patch( + "sds_gateway.api_methods.federation.reindex.fed_doc_exists", + new=MagicMock(return_value=True), + ) + def test_disconnect_captures_reindexes_orphans( + self, + mock_indexer_cls: MagicMock, + ) -> None: + mock_indexer = mock_indexer_cls.return_value + dataset = DatasetFactory(status=DatasetStatus.FINAL, is_public=True) + capture = CaptureFactory(is_deleted=False) + capture.datasets.add(dataset) + + with _federation_on_commit(): + disconnect_captures_from_dataset(dataset) + + capture.refresh_from_db() + assert not capture.datasets.exists() + mock_indexer.apply_local_event.assert_called_once() + body = mock_indexer.apply_local_event.call_args.kwargs["body"] + assert body["public_dataset_ids"] == [] + + @patch("sds_gateway.api_methods.federation.reindex.reindex_federated_asset") + @patch( + "sds_gateway.api_methods.federation.reindex.fed_doc_exists", + new=MagicMock(return_value=True), + ) + def test_dataset_soft_delete_reindexes_orphan_capture( + self, + mock_reindex: MagicMock, + ) -> None: + dataset = DatasetFactory(status=DatasetStatus.FINAL, is_public=True) + capture = CaptureFactory(is_deleted=False) + capture.datasets.add(dataset) + + with _federation_on_commit(): + dataset.soft_delete() + + capture.refresh_from_db() + assert capture.datasets.federation_exportable().exists() is False + capture_calls = [ + c + for c in mock_reindex.call_args_list + if c.args[1] == ItemType.CAPTURE and c.args[0].pk == capture.pk + ] + assert capture_calls + + +@override_settings( + FEDERATION_ENABLED=True, + FEDERATION_SITE_NAME="crc", + FEDERATION_OPERATIONAL_OVERRIDE=True, +) +class TestFederationReindexOnCommit(TestCase): + @patch("sds_gateway.api_methods.federation.reindex.reindex_federated_asset") + def test_rollback_skips_federation_reindex( + self, + mock_reindex: MagicMock, + ) -> None: + dataset = DatasetFactory(status=DatasetStatus.FINAL, is_public=True) + try: + with transaction.atomic(): + dataset.name = "rolled back" + dataset.save() + abort_msg = "abort" + raise RuntimeError(abort_msg) # noqa: TRY301 + except RuntimeError: + pass + mock_reindex.assert_not_called() diff --git a/gateway/sds_gateway/api_methods/tests/test_federation_sync_api_key.py b/gateway/sds_gateway/api_methods/tests/test_federation_sync_api_key.py new file mode 100644 index 000000000..47359baed --- /dev/null +++ b/gateway/sds_gateway/api_methods/tests/test_federation_sync_api_key.py @@ -0,0 +1,74 @@ +"""Tests for federation sync Token + mint endpoint.""" + +from __future__ import annotations + +import pytest +from django.conf import settings +from django.urls import reverse +from rest_framework import status +from rest_framework.authtoken.models import Token +from rest_framework.test import APIClient + +from sds_gateway.api_methods.models import KeySources +from sds_gateway.users.backend_service_key_utils import mint_federation_sync_api_key +from sds_gateway.users.backend_service_key_utils import update_service_server_token +from sds_gateway.users.models import User +from sds_gateway.users.models import UserAPIKey + +pytestmark = pytest.mark.django_db + + +class TestFederationSyncMint: + def setup_method(self) -> None: + email = settings.FEDERATION_SYNC_USER_EMAIL + User.objects.filter(email=email).delete() + update_service_server_token( + user_email=email, + token_key="a" * 40, + ) + + def test_mint_creates_api_key(self) -> None: + user = User.objects.get(email=settings.FEDERATION_SYNC_USER_EMAIL) + raw = mint_federation_sync_api_key(user) + assert raw is not None + key = UserAPIKey.objects.get_from_key(raw) + assert key.source == KeySources.FederationSync + assert key.user_id == user.pk + + def test_mint_replaces_prior_key_like_svi(self) -> None: + user = User.objects.get(email=settings.FEDERATION_SYNC_USER_EMAIL) + first = mint_federation_sync_api_key(user) + second = mint_federation_sync_api_key(user) + assert second != first + assert ( + UserAPIKey.objects.filter( + user=user, + source=KeySources.FederationSync, + ).count() + == 1 + ) + + +class TestGetFederationSyncApiKeyEndpoint: + def setup_method(self) -> None: + email = settings.FEDERATION_SYNC_USER_EMAIL + User.objects.filter(email=email).delete() + update_service_server_token( + user_email=email, + token_key="b" * 40, + ) + + def test_mint_via_http(self) -> None: + sync_email = settings.FEDERATION_SYNC_USER_EMAIL + sync_user = User.objects.get(email=sync_email) + target = User.objects.create(email="mint-target@example.com", is_approved=True) + token = Token.objects.get(user=sync_user) + client = APIClient() + url = reverse("users:get_federation_sync_api_key") + response = client.get( + f"{url}?email={target.email}", + HTTP_AUTHORIZATION=f"Token {token.key}", + ) + assert response.status_code == status.HTTP_200_OK + assert "api_key" in response.json() + assert response.json()["email"] == target.email diff --git a/gateway/sds_gateway/api_methods/tests/test_file_endpoints.py b/gateway/sds_gateway/api_methods/tests/test_file_endpoints.py index 7ee5add9f..afa6218e6 100644 --- a/gateway/sds_gateway/api_methods/tests/test_file_endpoints.py +++ b/gateway/sds_gateway/api_methods/tests/test_file_endpoints.py @@ -21,6 +21,7 @@ from sds_gateway.api_methods.models import Capture from sds_gateway.api_methods.models import CaptureType +from sds_gateway.api_methods.models import DatasetStatus from sds_gateway.api_methods.models import File from sds_gateway.api_methods.models import ItemType from sds_gateway.api_methods.serializers.file_serializers import FilePostSerializer @@ -680,7 +681,7 @@ def test_disabled_share_permission_blocks_file_access(self) -> None: ) def test_detach_file_from_datasets(self) -> None: - """PUT detach-from-datasets clears FK and M2M dataset links.""" + """PUT detach-from-datasets clears draft/private FK and M2M dataset links.""" dataset = DatasetFactory(owner=self.user) self.file.datasets.add(dataset) self.file.dataset = dataset @@ -691,8 +692,31 @@ def test_detach_file_from_datasets(self) -> None: assert response.status_code == status.HTTP_200_OK assert response.json()["message"] == ( - "File detached from all connected datasets successfully" + "File detached from draft/private datasets successfully" ) self.file.refresh_from_db() assert self.file.dataset is None assert not self.file.datasets.exists() + + def test_detach_file_keeps_published_dataset_links(self) -> None: + draft = DatasetFactory( + owner=self.user, status=DatasetStatus.DRAFT, is_public=False + ) + published = DatasetFactory( + owner=self.user, + status=DatasetStatus.FINAL, + is_public=True, + ) + self.file.datasets.add(draft, published) + self.file.dataset = draft + self.file.save() + + url = reverse("api:files-detach-from-datasets", args=[str(self.file.uuid)]) + response = self.client.put(url) + + assert response.status_code == status.HTTP_200_OK + self.file.refresh_from_db() + assert list(self.file.datasets.values_list("uuid", flat=True)) == [ + published.uuid + ] + assert self.file.dataset is None diff --git a/gateway/sds_gateway/api_methods/utils/asset_access_control.py b/gateway/sds_gateway/api_methods/utils/asset_access_control.py index 3e1795f1d..56ce8e423 100644 --- a/gateway/sds_gateway/api_methods/utils/asset_access_control.py +++ b/gateway/sds_gateway/api_methods/utils/asset_access_control.py @@ -388,7 +388,18 @@ def disconnect_captures_from_dataset(dataset: Dataset) -> None: Clears ``Capture.datasets`` (M2M) and deprecated ``Capture.dataset`` (FK). """ + from sds_gateway.api_methods.federation.reindex import ( # noqa: PLC0415 + reindex_captures_after_dataset_unlink, + ) + + capture_pks = list( + Capture.datasets.through.objects.filter(dataset_id=dataset.pk).values_list( + "capture_id", + flat=True, + ), + ) Capture.datasets.through.objects.filter(dataset_id=dataset.pk).delete() + reindex_captures_after_dataset_unlink(capture_pks) # TODO: remove FK after contraction Capture.objects.filter(dataset=dataset).update(dataset=None) diff --git a/gateway/sds_gateway/api_methods/utils/relationship_utils.py b/gateway/sds_gateway/api_methods/utils/relationship_utils.py index 02dc5c727..b8041df5f 100644 --- a/gateway/sds_gateway/api_methods/utils/relationship_utils.py +++ b/gateway/sds_gateway/api_methods/utils/relationship_utils.py @@ -9,10 +9,12 @@ """ from django.db import transaction +from django.db.models import Q from django.db.models import QuerySet from sds_gateway.api_methods.models import Capture from sds_gateway.api_methods.models import Dataset +from sds_gateway.api_methods.models import DatasetStatus from sds_gateway.api_methods.models import File @@ -305,12 +307,27 @@ def get_capture_datasets( return union_to_queryset(datasets_union, Dataset) +def _datasets_removable_from_item(item: Capture | File) -> QuerySet[Dataset]: + if isinstance(item, Capture): + linked = get_capture_datasets(item, include_deleted=False) + else: + linked = get_file_datasets(item, include_deleted=False) + return linked.exclude(Q(status=DatasetStatus.FINAL) | Q(is_public=True)) + + @transaction.atomic def detach_item_from_all_datasets(item: Capture | File) -> None: - """Clear deprecated dataset FK and M2M links for a single capture or file.""" - item.dataset = None - item.datasets.clear() - item.save() + """Remove draft/private dataset links; keep published (FINAL or public) links.""" + to_remove = _datasets_removable_from_item(item) + remove_pks = list(to_remove.values_list("pk", flat=True)) + if remove_pks: + item.datasets.remove(*remove_pks) + + fk_dataset = item.dataset + if fk_dataset is not None and not fk_dataset.is_deleted: + if fk_dataset.status != DatasetStatus.FINAL and not fk_dataset.is_public: + item.dataset = None + item.save(update_fields=["dataset"]) def group_captures_by_top_level_dir( diff --git a/gateway/sds_gateway/api_methods/views/capture_endpoints.py b/gateway/sds_gateway/api_methods/views/capture_endpoints.py index 94f23808e..781a69dab 100644 --- a/gateway/sds_gateway/api_methods/views/capture_endpoints.py +++ b/gateway/sds_gateway/api_methods/views/capture_endpoints.py @@ -27,7 +27,6 @@ from rest_framework.authentication import SessionAuthentication from rest_framework.decorators import action from rest_framework.pagination import PageNumberPagination -from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response @@ -208,7 +207,6 @@ class CapturePagination(PageNumberPagination): class CaptureViewSet(viewsets.ViewSet): authentication_classes = [SessionAuthentication, APIKeyAuthentication] - permission_classes = [IsAuthenticated] def _validate_and_index_metadata( self, @@ -1079,16 +1077,19 @@ def revoke_share_permissions( ], responses={ 200: OpenApiResponse( - description="Capture detached from all datasets successfully" + description="Capture detached from draft/private datasets successfully" ), 400: OpenApiResponse(description="Bad Request"), 500: OpenApiResponse(description="Internal Server Error"), }, - description="Detach a capture from all connected datasets.", - summary="Detach Capture from All Connected Datasets", + description=( + "Detach a capture from draft and private datasets. " + "Published datasets (final or public) stay linked." + ), + summary="Detach Capture from Draft/Private Datasets", ) def detach_from_datasets(self, request: Request, pk: str | None = None) -> Response: - """Detach a capture from all datasets.""" + """Detach a capture from draft and private datasets (not published).""" if pk is None: return Response( {"detail": "Capture UUID is required."}, @@ -1102,7 +1103,9 @@ def detach_from_datasets(self, request: Request, pk: str | None = None) -> Respo return Response( status=status.HTTP_200_OK, data={ - "message": "Capture detached from all connected datasets successfully", + "message": ( + "Capture detached from draft/private datasets successfully" + ), }, ) diff --git a/gateway/sds_gateway/api_methods/views/dataset_endpoints.py b/gateway/sds_gateway/api_methods/views/dataset_endpoints.py index 0785ece85..2ef78566f 100644 --- a/gateway/sds_gateway/api_methods/views/dataset_endpoints.py +++ b/gateway/sds_gateway/api_methods/views/dataset_endpoints.py @@ -11,7 +11,6 @@ from rest_framework import status from rest_framework.authentication import SessionAuthentication from rest_framework.decorators import action -from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework.viewsets import ViewSet @@ -56,8 +55,7 @@ def _truthy_query_param(raw: str | None) -> bool: class DatasetViewSet(ViewSet): - authentication_classes = [SessionAuthentication, APIKeyAuthentication] - permission_classes = [IsAuthenticated] + authentication_classes = [APIKeyAuthentication, SessionAuthentication] def _get_file_objects( self, diff --git a/gateway/sds_gateway/api_methods/views/federation_endpoints.py b/gateway/sds_gateway/api_methods/views/federation_endpoints.py new file mode 100644 index 000000000..f51816552 --- /dev/null +++ b/gateway/sds_gateway/api_methods/views/federation_endpoints.py @@ -0,0 +1,111 @@ +"""Federation export API (sync service reads local fed-* OpenSearch docs).""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from uuid import UUID + +from drf_spectacular.utils import extend_schema +from rest_framework.decorators import action +from rest_framework.exceptions import NotFound +from rest_framework.response import Response +from rest_framework.viewsets import ViewSet + +if TYPE_CHECKING: + from rest_framework.request import Request + +from sds_gateway.api_methods.authentication import APIKeyAuthentication +from sds_gateway.api_methods.federation.compile_federated_data import ( + compile_federated_capture_doc, +) +from sds_gateway.api_methods.federation.compile_federated_data import ( + compile_federated_dataset_doc, +) +from sds_gateway.api_methods.federation.compile_federated_data import ( + get_federated_export_doc_by_uuid, +) +from sds_gateway.api_methods.federation.compile_federated_data import ( + public_captures_queryset, +) +from sds_gateway.api_methods.federation.compile_federated_data import ( + public_datasets_queryset, +) +from sds_gateway.api_methods.federation.permissions import ( + IsFederationInternalExportClient, +) +from sds_gateway.api_methods.federation.permissions import IsFederationOperational +from sds_gateway.api_methods.models import ItemType +from sds_gateway.api_methods.permissions import IsFederationSyncKey + + +def _federation_export_uuid(pk: str | None) -> UUID: + if not pk: + raise NotFound + try: + return UUID(pk) + except ValueError as exc: + raise NotFound from exc + + +@extend_schema(exclude=True) +class FederationViewSet(ViewSet): + """Internal export endpoints for the federation sync service.""" + + authentication_classes = [APIKeyAuthentication] + permission_classes = [ + IsFederationSyncKey, + IsFederationOperational, + IsFederationInternalExportClient, + ] + + @action(detail=False, methods=["get"], url_path="export/datasets") + def export_datasets_list(self, request: Request) -> Response: + """List all public finalized datasets for federation bootstrap.""" + datasets = public_datasets_queryset() + return Response( + [compile_federated_dataset_doc(dataset) for dataset in datasets], + ) + + @action( + detail=False, + methods=["get"], + url_path=r"export/datasets/(?P[^/.]+)", + ) + def export_dataset_detail( + self, request: Request, pk: str | None = None + ) -> Response: + """Return one indexed dataset doc (post-signal); 404 if not in fed-datasets.""" + asset_uuid = _federation_export_uuid(pk) + body = get_federated_export_doc_by_uuid( + asset_type=ItemType.DATASET, + asset_uuid=asset_uuid, + ) + if body is None: + raise NotFound + return Response(body) + + @action(detail=False, methods=["get"], url_path="export/captures") + def export_captures_list(self, request: Request) -> Response: + """List all public captures for federation bootstrap.""" + captures = public_captures_queryset() + return Response( + [compile_federated_capture_doc(capture) for capture in captures], + ) + + @action( + detail=False, + methods=["get"], + url_path=r"export/captures/(?P[^/.]+)", + ) + def export_capture_detail( + self, request: Request, pk: str | None = None + ) -> Response: + """Return one indexed capture doc (post-signal); 404 if not in fed-captures.""" + asset_uuid = _federation_export_uuid(pk) + body = get_federated_export_doc_by_uuid( + asset_type=ItemType.CAPTURE, + asset_uuid=asset_uuid, + ) + if body is None: + raise NotFound + return Response(body) diff --git a/gateway/sds_gateway/api_methods/views/file_endpoints.py b/gateway/sds_gateway/api_methods/views/file_endpoints.py index 0a441566b..6b5444afc 100644 --- a/gateway/sds_gateway/api_methods/views/file_endpoints.py +++ b/gateway/sds_gateway/api_methods/views/file_endpoints.py @@ -24,7 +24,6 @@ from rest_framework import status from rest_framework.decorators import action from rest_framework.pagination import PageNumberPagination -from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from rest_framework.views import APIView @@ -65,7 +64,6 @@ class FilePagination(PageNumberPagination): class FileViewSet(ViewSet): authentication_classes = [APIKeyAuthentication] - permission_classes = [IsAuthenticated] @staticmethod def _paginated_list_response( @@ -561,16 +559,19 @@ def update(self, request: Request, pk: str | None = None) -> Response: ], responses={ 200: OpenApiResponse( - description="File detached from all datasets successfully" + description="File detached from draft/private datasets successfully" ), 400: OpenApiResponse(description="Bad Request"), 500: OpenApiResponse(description="Internal Server Error"), }, - description="Detach a file from all datasets.", - summary="Detach File from All Connected Datasets", + description=( + "Detach a file from draft and private datasets. " + "Published datasets (final or public) stay linked." + ), + summary="Detach File from Draft/Private Datasets", ) def detach_from_datasets(self, request: Request, pk: str | None = None) -> Response: - """Detach a file from all datasets.""" + """Detach a file from draft and private datasets (not published).""" if pk is None: return Response( {"detail": "File UUID is required."}, @@ -584,7 +585,7 @@ def detach_from_datasets(self, request: Request, pk: str | None = None) -> Respo return Response( status=status.HTTP_200_OK, data={ - "message": "File detached from all connected datasets successfully", + "message": "File detached from draft/private datasets successfully", }, ) @@ -698,7 +699,6 @@ def download_file(self, request: Request, pk: str | None = None) -> HttpResponse class CheckFileContentsExistView(APIView): authentication_classes = [APIKeyAuthentication] - permission_classes = [IsAuthenticated] @extend_schema( request=FilePostSerializer, diff --git a/gateway/sds_gateway/users/api/views.py b/gateway/sds_gateway/users/api/views.py index 1555e5ad2..c4b288b09 100644 --- a/gateway/sds_gateway/users/api/views.py +++ b/gateway/sds_gateway/users/api/views.py @@ -1,3 +1,4 @@ +from collections.abc import Callable from typing import cast from django.conf import settings @@ -18,9 +19,9 @@ from rest_framework.views import APIView from rest_framework.viewsets import GenericViewSet -from sds_gateway.api_methods.models import KeySources +from sds_gateway.users.backend_service_key_utils import mint_federation_sync_api_key +from sds_gateway.users.backend_service_key_utils import mint_svi_backend_api_key from sds_gateway.users.models import User -from sds_gateway.users.models import UserAPIKey from .serializers import UserSerializer @@ -49,17 +50,21 @@ def me(self, request: HttpRequest) -> Response: @extend_schema(exclude=True) -class GetAPIKeyView(APIView): +class BackendServiceMintAPIKeyView(APIView): + """Token-authenticated mint endpoint for internal backend services.""" + permission_classes = [IsAuthenticated] authentication_classes = [TokenAuthentication] + service_user_email_setting = "" + unauthorized_message = "" + mint_user_api_key: Callable[[User], str] | None = None def get(self, request: Request) -> Response: - """Generates an API Key for an SDS user.""" - + allowed_email = getattr(settings, self.service_user_email_setting) log.debug(f"request.user: {request.user}") - if request.user.email != settings.SVI_SERVER_EMAIL: + if request.user.email != allowed_email: return Response( - {"error": "Unauthorized. Only SVI Server can access this endpoint."}, + {"error": self.unauthorized_message}, status=status.HTTP_403_FORBIDDEN, ) @@ -70,17 +75,21 @@ def get(self, request: Request) -> Response: status=status.HTTP_400_BAD_REQUEST, ) - try: - user = get_object_or_404(User, email=email, is_approved=True) - UserAPIKey.objects.filter(user=user, source=KeySources.SVIBackend).delete() - _, raw_key = UserAPIKey.objects.create_key( - name=f"{user.email}-SVI-API-KEY", - user=user, - source=KeySources.SVIBackend, - ) - return Response({"api_key": raw_key, "email": user.email}) - except User.DoesNotExist: - return Response( - {"error": "User not found"}, - status=status.HTTP_404_NOT_FOUND, - ) + user = get_object_or_404(User, email=email, is_approved=True) + raw_key = self.mint_user_api_key(user) + return Response({"api_key": raw_key, "email": user.email}) + + +get_svi_api_key_view = BackendServiceMintAPIKeyView.as_view( + service_user_email_setting="SVI_SERVER_EMAIL", + unauthorized_message="Unauthorized. Only SVI Server can access this endpoint.", + mint_user_api_key=mint_svi_backend_api_key, +) + +get_federation_sync_api_key_view = BackendServiceMintAPIKeyView.as_view( + service_user_email_setting="FEDERATION_SYNC_USER_EMAIL", + unauthorized_message=( + "Unauthorized. Only federation sync can access this endpoint." + ), + mint_user_api_key=mint_federation_sync_api_key, +) diff --git a/gateway/sds_gateway/users/backend_service_key_utils.py b/gateway/sds_gateway/users/backend_service_key_utils.py new file mode 100644 index 000000000..167ff217e --- /dev/null +++ b/gateway/sds_gateway/users/backend_service_key_utils.py @@ -0,0 +1,109 @@ +"""Internal service auth: DRF Tokens and UserAPIKey mint (SVI, federation sync).""" + +from __future__ import annotations + +import logging + +from django.conf import settings +from django.core.exceptions import ValidationError +from django.db import DatabaseError +from django.db import IntegrityError +from rest_framework.authtoken.models import Token + +from sds_gateway.api_methods.models import KeySources +from sds_gateway.users.models import User +from sds_gateway.users.models import UserAPIKey + +logger = logging.getLogger(__name__) + + +def get_or_create_service_user(email: str) -> tuple[User, bool]: + user, created = User.objects.get_or_create( + email=email, + defaults={ + "is_active": True, + "is_approved": True, + }, + ) + if created: + user.set_unusable_password() + user.save(update_fields=["password"]) + logger.info("Created service user: %s", email) + return user, created + + +def update_service_server_token(*, user_email: str, token_key: str) -> None: + """Sync DRF authtoken ``Token.key`` for an internal service user.""" + user, _created = get_or_create_service_user(user_email) + existing = Token.objects.filter(user=user) + count = existing.count() + if count: + existing.delete() + logger.info("Removed %d token(s) for service user %s", count, user_email) + Token.objects.create(user=user, key=token_key) + logger.info("Updated service token for %s", user_email) + + +def replace_user_api_key_for_source( + *, + user: User, + source: KeySources, + name: str, + description: str = "", +) -> str: + """Delete existing keys for ``(user, source)`` and mint a new Api-Key.""" + UserAPIKey.objects.filter(user=user, source=source).delete() + _obj, raw_key = UserAPIKey.objects.create_key( + name=name, + user=user, + source=source, + description=description, + ) + return raw_key + + +def mint_svi_backend_api_key(user: User) -> str: + return replace_user_api_key_for_source( + user=user, + source=KeySources.SVIBackend, + name=f"{user.email}-SVI-API-KEY", + ) + + +def mint_federation_sync_api_key(user: User) -> str: + return replace_user_api_key_for_source( + user=user, + source=KeySources.FederationSync, + name="federation-sync", + description="Federation sync service (export endpoints only)", + ) + + +def update_svi_server_token() -> None: + """Update the SVI token when ``SVI_SERVER_API_KEY`` changes.""" + try: + update_service_server_token( + user_email=settings.SVI_SERVER_EMAIL, + token_key=settings.SVI_SERVER_API_KEY, + ) + except (DatabaseError, IntegrityError, ValidationError): + logger.exception("Failed to update SVI server token") + + +def update_federation_sync_drf_token() -> None: + """Sync federation-sync DRF ``Token`` from ``FEDERATION_SYNC_DRF_TOKEN``.""" + token_key = settings.FEDERATION_SYNC_DRF_TOKEN + if len(token_key) != settings.TOKEN_LENGTH: + msg = ( + f"FEDERATION_SYNC_DRF_TOKEN must be {settings.TOKEN_LENGTH} characters " + f"(load federation-shared.env on the gateway container)" + ) + raise ValueError(msg) + try: + update_service_server_token( + user_email=settings.FEDERATION_SYNC_USER_EMAIL, + token_key=token_key, + ) + except (DatabaseError, IntegrityError, ValidationError): + logger.exception("Failed to update federation sync DRF token") + raise diff --git a/gateway/sds_gateway/users/management/commands/create_test_files.py b/gateway/sds_gateway/users/management/commands/create_test_files.py deleted file mode 100644 index 9c6f0b553..000000000 --- a/gateway/sds_gateway/users/management/commands/create_test_files.py +++ /dev/null @@ -1,140 +0,0 @@ -import secrets -from datetime import timedelta -from pathlib import Path - -from django.conf import settings -from django.contrib.auth import get_user_model -from django.core.management.base import BaseCommand -from django.db import DatabaseError -from django.db import IntegrityError -from django.db import transaction -from django.utils import timezone - -from sds_gateway.captures.models import Capture -from sds_gateway.files.models import File -from sds_gateway.minio_client import MinioClient - -User = get_user_model() # This is the only User import we need - - -class Command(BaseCommand): - help = "Create test files and captures for development" - - def handle(self, *args, **options): - try: - if not settings.DEBUG: - self.stdout.write( - self.style.ERROR( - "This command can only be run in local development environment" - ) - ) - return - user = self._get_or_create_test_user() - self._create_test_files(user) - self._create_test_captures(user) - except (DatabaseError, IntegrityError) as e: - self.stdout.write(self.style.ERROR(f"Database error: {e}")) - except OSError as e: - self.stdout.write(self.style.ERROR(f"I/O error: {e}")) - - def _get_or_create_test_user(self): - """Get or create a test user.""" - try: - return User.objects.get_or_create( - email="test@example.com", - defaults={ - "name": "Test User", - "is_staff": True, - "is_superuser": True, - }, - )[0] - except DatabaseError as e: - self.stdout.write(self.style.ERROR(f"Failed to create test user: {e}")) - raise - - def _create_test_files(self, user): - """Create test files in MinIO.""" - # Define test directories and file types - directories = [Path(f"files/{user.email}/test_dir_{i}") for i in range(3)] - - file_types = [ - ("txt", b"Test content"), - ("csv", b"id,name\n1,test"), - ("json", b'{"test": "data"}'), - ] - - # Create directories if they don't exist - for directory in directories: - directory_path = Path(str(directory)) - directory_path.mkdir(parents=True, exist_ok=True) - - # Create files - minio_client = MinioClient() - created_files = [] - - try: - for directory in directories: - for i in range(3): - file_type, content = secrets.choice(file_types) - file_name = f"test_file_{i}.{file_type}" - file_path = directory / file_name - - # Save file - with file_path.open("wb") as f: - f.write(content) - - # Upload to MinIO - minio_client.upload_file( - str(file_path), - str(file_path), - f"text/{file_type}", - ) - - # Create file record - file_obj = File.objects.create( - name=file_name, - directory=str(directory), - media_type=f"text/{file_type}", - size=len(content), - owner=user, - ) - created_files.append(file_obj) - - self.stdout.write(self.style.SUCCESS(f"Created file: {file_path}")) - except (OSError, DatabaseError) as e: - self.stdout.write(self.style.ERROR(f"Failed to create test files: {e}")) - raise - else: - return created_files - - def _create_test_captures(self, user): - """Create test captures.""" - capture_types = ["rh", "drf"] - channels = ["ch1", "ch2", "ch3"] - scan_groups = ["group1", "group2"] - - try: - with transaction.atomic(): - for i in range(5): - capture_type = secrets.choice(capture_types) - channel = secrets.choice(channels) - scan_group = secrets.choice(scan_groups) - top_level_dir = f"captures/{user.email}/test_capture_{i}" - - capture = Capture.objects.create( - capture_type=capture_type, - top_level_dir=top_level_dir, - channel=channel, - scan_group=scan_group, - owner=user, - created_at=timezone.now() - timedelta(days=i), - ) - - # Associate some random files with the capture - associated_files = File.objects.filter(owner=user).order_by("?")[:3] - capture.files.set(associated_files) - - self.stdout.write(self.style.SUCCESS(f"Created capture: {capture}")) - - except (DatabaseError, IntegrityError) as e: - self.stdout.write(self.style.WARNING(f"Failed to create capture {i}: {e}")) diff --git a/gateway/sds_gateway/users/management/commands/init_federation_sync_token.py b/gateway/sds_gateway/users/management/commands/init_federation_sync_token.py new file mode 100644 index 000000000..48e5d0e7b --- /dev/null +++ b/gateway/sds_gateway/users/management/commands/init_federation_sync_token.py @@ -0,0 +1,21 @@ +"""Management command to initialize federation sync server token.""" + +from django.core.management.base import BaseCommand +from loguru import logger as log + +from sds_gateway.users.backend_service_key_utils import update_federation_sync_drf_token + + +class Command(BaseCommand): + help = "Initialize or update the federation sync server authentication token" + + def handle(self, *args, **options) -> None: + """Initialize the federation sync server token.""" + log.info("Initializing federation sync server token...") + + try: + update_federation_sync_drf_token() + log.success("Federation sync server token initialized successfully") + except Exception as e: + log.error(f"Failed to initialize federation sync server token: {e}") + raise diff --git a/gateway/sds_gateway/users/management/commands/init_svi_token.py b/gateway/sds_gateway/users/management/commands/init_svi_token.py index a28bf690a..64007130d 100644 --- a/gateway/sds_gateway/users/management/commands/init_svi_token.py +++ b/gateway/sds_gateway/users/management/commands/init_svi_token.py @@ -3,7 +3,7 @@ from django.core.management.base import BaseCommand from loguru import logger as log -from sds_gateway.users.svi_utils import update_svi_server_token +from sds_gateway.users.backend_service_key_utils import update_svi_server_token class Command(BaseCommand): diff --git a/gateway/sds_gateway/users/migrations/0012_alter_userapikey_source_federation_sync.py b/gateway/sds_gateway/users/migrations/0012_alter_userapikey_source_federation_sync.py new file mode 100644 index 000000000..084b10c0b --- /dev/null +++ b/gateway/sds_gateway/users/migrations/0012_alter_userapikey_source_federation_sync.py @@ -0,0 +1,38 @@ +# Generated manually for FederationSync API key source + +from django.db import migrations, models + +import sds_gateway.api_methods.models + + +class Migration(migrations.Migration): + + dependencies = [ + ("users", "0011_user_orcid_id_alter_user_is_approved"), + ] + + operations = [ + migrations.AlterField( + model_name="userapikey", + name="source", + field=models.CharField( + choices=[ + (sds_gateway.api_methods.models.KeySources["SDSWebUI"], "SDS Web UI"), + ( + sds_gateway.api_methods.models.KeySources["SVIBackend"], + "SVI Backend", + ), + ( + sds_gateway.api_methods.models.KeySources["SVIWebUI"], + "SVI Web UI", + ), + ( + sds_gateway.api_methods.models.KeySources["FederationSync"], + "Federation Sync", + ), + ], + default=sds_gateway.api_methods.models.KeySources["SDSWebUI"], + max_length=255, + ), + ), + ] diff --git a/gateway/sds_gateway/users/migrations/max_migration.txt b/gateway/sds_gateway/users/migrations/max_migration.txt index fda121cc0..7707335c4 100644 --- a/gateway/sds_gateway/users/migrations/max_migration.txt +++ b/gateway/sds_gateway/users/migrations/max_migration.txt @@ -1 +1 @@ -0011_user_orcid_id_alter_user_is_approved +0012_alter_userapikey_source_federation_sync \ No newline at end of file diff --git a/gateway/sds_gateway/users/models.py b/gateway/sds_gateway/users/models.py index b1a31d0db..f5b23370f 100644 --- a/gateway/sds_gateway/users/models.py +++ b/gateway/sds_gateway/users/models.py @@ -64,6 +64,7 @@ class UserAPIKey(AbstractAPIKey): (KeySources.SDSWebUI, "SDS Web UI"), (KeySources.SVIBackend, "SVI Backend"), (KeySources.SVIWebUI, "SVI Web UI"), + (KeySources.FederationSync, "Federation Sync"), ] user = cast( "User", diff --git a/gateway/sds_gateway/users/svi_utils.py b/gateway/sds_gateway/users/svi_utils.py deleted file mode 100644 index 40273a0e9..000000000 --- a/gateway/sds_gateway/users/svi_utils.py +++ /dev/null @@ -1,53 +0,0 @@ -"""Utilities for managing SVI server authentication.""" - -from __future__ import annotations - -import logging - -from django.conf import settings -from django.core.exceptions import ValidationError -from django.db import DatabaseError -from django.db import IntegrityError -from rest_framework.authtoken.models import Token - -from sds_gateway.users.models import User - -logger = logging.getLogger(__name__) - - -def update_svi_server_token() -> None: - """Updates the SVI's token for automatic key rotation if the setting changes.""" - try: - svi_user, created = User.objects.get_or_create( - email=settings.SVI_SERVER_EMAIL, - defaults={ - "is_active": True, - "is_approved": True, - }, - ) - - if created: - logger.info("Created SVI server user: %s", settings.SVI_SERVER_EMAIL) - - __delete_svi_user_tokens(svi_user) - __create_svi_user_token(svi_user) - - except (DatabaseError, IntegrityError, ValidationError): - logger.exception("Failed to update SVI server token") - - -def __delete_svi_user_tokens(svi_user: User) -> None: - """Deletes all existing tokens for the SVI user, revoking existing keys.""" - existing_tokens = Token.objects.filter(user=svi_user) - token_count = existing_tokens.count() - if token_count > 0: - existing_tokens.delete() - logger.info("Removed %d existing token(s) for SVI server user", token_count) - - -def __create_svi_user_token(svi_user: User) -> None: - """Creates a new token for the SVI user following the active setting.""" - Token.objects.create(user=svi_user, key=settings.SVI_SERVER_API_KEY) - logger.info( - "Successfully updated SVI server token for %s", settings.SVI_SERVER_EMAIL - ) diff --git a/gateway/sds_gateway/users/tests/test_federation_sync_drf_init.py b/gateway/sds_gateway/users/tests/test_federation_sync_drf_init.py new file mode 100644 index 000000000..bac764ccb --- /dev/null +++ b/gateway/sds_gateway/users/tests/test_federation_sync_drf_init.py @@ -0,0 +1,30 @@ +"""Tests for federation sync DRF token init from shared env.""" + +from __future__ import annotations + +import pytest +from django.conf import settings +from django.test import override_settings +from rest_framework.authtoken.models import Token + +from sds_gateway.users.backend_service_key_utils import update_federation_sync_drf_token +from sds_gateway.users.models import User + +pytestmark = pytest.mark.django_db + + +@override_settings(FEDERATION_SYNC_DRF_TOKEN="a" * 40) +def test_update_federation_sync_drf_token_from_settings() -> None: + email = settings.FEDERATION_SYNC_USER_EMAIL + User.objects.filter(email=email).delete() + + update_federation_sync_drf_token() + + user = User.objects.get(email=email) + assert Token.objects.get(user=user).key == "a" * 40 + + +@override_settings(FEDERATION_SYNC_DRF_TOKEN="") +def test_update_requires_token_length() -> None: + with pytest.raises(ValueError, match="FEDERATION_SYNC_DRF_TOKEN"): + update_federation_sync_drf_token() diff --git a/gateway/sds_gateway/users/tests/test_quick_add_views.py b/gateway/sds_gateway/users/tests/test_quick_add_views.py index 5526a0918..76bd8712b 100644 --- a/gateway/sds_gateway/users/tests/test_quick_add_views.py +++ b/gateway/sds_gateway/users/tests/test_quick_add_views.py @@ -11,6 +11,7 @@ from sds_gateway.api_methods.models import Capture from sds_gateway.api_methods.models import CaptureType +from sds_gateway.api_methods.models import Dataset from sds_gateway.api_methods.models import DatasetStatus from sds_gateway.api_methods.models import ItemType from sds_gateway.api_methods.models import PermissionLevel @@ -509,7 +510,9 @@ def test_response_contains_uuid_and_name(self, client: Client, owner: User): assert entry["name"] == "My Dataset" def test_unnamed_dataset_shows_fallback(self, client: Client, owner: User): - ds = DatasetFactory(owner=owner, is_public=False, name="", keywords=None) + ds = DatasetFactory(owner=owner, is_public=False, keywords=None) + Dataset.objects.filter(pk=ds.pk).update(name="") + ds.refresh_from_db() client.force_login(owner) response = self._get(client) diff --git a/gateway/sds_gateway/users/tests/test_svi_utils.py b/gateway/sds_gateway/users/tests/test_svi_utils.py index 8af126081..e79e97aa2 100644 --- a/gateway/sds_gateway/users/tests/test_svi_utils.py +++ b/gateway/sds_gateway/users/tests/test_svi_utils.py @@ -5,8 +5,8 @@ from django.db.utils import IntegrityError from rest_framework.authtoken.models import Token +from sds_gateway.users.backend_service_key_utils import update_svi_server_token from sds_gateway.users.models import User -from sds_gateway.users.svi_utils import update_svi_server_token @pytest.mark.django_db diff --git a/gateway/sds_gateway/users/urls.py b/gateway/sds_gateway/users/urls.py index 635241e9a..4443b4c40 100644 --- a/gateway/sds_gateway/users/urls.py +++ b/gateway/sds_gateway/users/urls.py @@ -1,7 +1,8 @@ from django.urls import path from django.views.generic import RedirectView -from .api.views import GetAPIKeyView +from .api.views import get_federation_sync_api_key_view +from .api.views import get_svi_api_key_view from .views import CheckFileExistsView from .views import FileContentView from .views import FileDownloadView @@ -129,7 +130,12 @@ path("upload-files/", UploadCaptureView.as_view(), name="upload_files"), path("check-file-exists/", CheckFileExistsView.as_view(), name="check_file_exists"), # Used by SVI Server to get API key for a user - path("get-svi-api-key/", GetAPIKeyView.as_view(), name="get_svi_api_key"), + path("get-svi-api-key/", get_svi_api_key_view, name="get_svi_api_key"), + path( + "get-federation-sync-api-key/", + get_federation_sync_api_key_view, + name="get_federation_sync_api_key", + ), path("revoke-api-key/", revoke_api_key_view, name="revoke_api_key"), path( "generate-api-key-form/",