Scaffolding: Federation gateway setup

gateway/.envs/example/django.env

@@ -8,6 +8,19 @@ HOSTNAME=localhost:8000
 API_VERSION=v1
 API_KEY=
 
+# FEDERATION
+# ------------------------------------------------------------------------------
+# Peer sync uses the federation-sync Docker service (same sds-network as gateway;
+# service definition lives under /federation). Bootstrap: enable federation, run
+# create_federation_sync_api_key, pass the key to federation-sync. Set FEDERATION_SITE_NAME
+# (e.g. crc) when enabling federation; use SDS_SITE_FQDN for the public host (RFC [site].fqdn).
+# FEDERATION_ENABLED=true  # Master switch for export APIs and Redis federation events.
+# FEDERATION_SITE_NAME=crc  # RFC [site].name (short peer id); set SDS_SITE_FQDN separately for [site].fqdn.
+# FEDERATION_EVENTS_CHANNEL=federation:events  # Redis pub/sub channel federation-sync subscribes to.
+# FEDERATION_SYNC_HEALTH_URL=http://federation-sync:8000/sync/health  # Health probe target (federation-sync service).
+# FEDERATION_SYNC_USER_EMAIL=federation-sync@internal.local  # Service user email for create_federation_sync_api_key.
+# FEDERATION_EXPORT_ALLOWED_CIDRS=  # Comma-separated CIDRs allowed to call export (default: private Docker ranges).
+
 # AUTH0
 # ------------------------------------------------------------------------------
 # Set these from your Auth0 application settings

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")

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
@@ -13,6 +14,9 @@
 from config.settings.logs import ColoredFormatter
 from config.settings.utils import guess_admin_console_env
 from config.settings.utils import guess_max_web_download_size
+from sds_gateway.api_methods.federation.redis_channel import (
+    resolve_federation_events_channel,
+)
 
 __rng = random.SystemRandom()
 
@@ -23,6 +27,8 @@ def __get_random_token(length: int) -> str:
         __rng.choice(string.ascii_letters + string.digits) for _ in range(length)
     )
 
+def _parse_cidrs(raw: list[str]) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
+    return [ipaddress.ip_network(item.strip(), strict=False) for item in raw]
 
 env.read_env()
 
@@ -610,7 +616,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 +718,63 @@ 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",
+)
+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
+] = _parse_cidrs(
+    env.list(
+        "FEDERATION_EXPORT_ALLOWED_CIDRS",
+        default=[
+            "127.0.0.1/32",
+            "::1/128",
+            "10.0.0.0/8",
+            "172.16.0.0/12",
+            "192.168.0.0/16",
+        ],
+    ),
+)
+
 # 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

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

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."

gateway/sds_gateway/api_methods/apps.py

@@ -12,7 +12,13 @@ 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,
+        )
+
+        initialize_federation_operational_state()
 
         silence_unwanted_logs()
 

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.
@@ -47,7 +47,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