From 663f8e75649d05a0ccf4341f843c85338697a2ac Mon Sep 17 00:00:00 2001 From: Alejandro Date: Thu, 16 Jul 2026 11:48:55 +0200 Subject: [PATCH 1/3] Add Inbox retention cleaner --- deploy/docker/.env.example | 6 + deploy/docker/Makefile | 7 + deploy/docker/README.md | 27 ++++ deploy/docker/docker-compose.yml | 28 ++++ src/handler/code/cleanup.py | 169 ++++++++++++++++++++++++ src/handler/code/handlers/accession.py | 39 +++++- src/handler/code/utils/db.py | 10 ++ src/vault/initdb.d/35-inbox-cleanup.sql | 55 ++++++++ src/vault/initdb.d/90-grants.sql | 4 +- 9 files changed, 343 insertions(+), 2 deletions(-) create mode 100644 src/handler/code/cleanup.py create mode 100644 src/vault/initdb.d/35-inbox-cleanup.sql diff --git a/deploy/docker/.env.example b/deploy/docker/.env.example index 89f7195a..87eadfc1 100644 --- a/deploy/docker/.env.example +++ b/deploy/docker/.env.example @@ -45,3 +45,9 @@ EGAF_COUNTER_START=0 SYNC_INTERVAL_SECONDS=60 EGA_PRECREATE_HOMES=0 LEGA_LOG=debug + +# Inbox cleanup is safe-by-default. Keep DRY_RUN=true until candidates and +# directory permissions have been checked in production. +INBOX_RETENTION_DAYS=90 +INBOX_CLEANUP_DRY_RUN=true +INBOX_CLEANUP_INTERVAL_HOURS=24 diff --git a/deploy/docker/Makefile b/deploy/docker/Makefile index d6b5c9c2..9d5cc817 100644 --- a/deploy/docker/Makefile +++ b/deploy/docker/Makefile @@ -1,5 +1,7 @@ SHELL := /bin/bash +DOCKER ?= podman + LEGA_UID := $(shell id -u lega) LEGA_GID := $(shell id -g lega) # LEGA_UID := $(shell id -u) @@ -59,6 +61,11 @@ export PGDATABASE=ega psql: psql +# Apply the Inbox cleanup table to an existing Vault DB. New databases load +# it automatically from src/vault/initdb.d/35-inbox-cleanup.sql. +apply-inbox-cleanup-migration: + $(DOCKER) exec -i vault-db psql -v ON_ERROR_STOP=1 -U postgres -d ega < ../../src/vault/initdb.d/35-inbox-cleanup.sql + ############################# ## Utility targets diff --git a/deploy/docker/README.md b/deploy/docker/README.md index 3483db58..49dd8d2c 100644 --- a/deploy/docker/README.md +++ b/deploy/docker/README.md @@ -134,4 +134,31 @@ and tear all down with docker-compose down -v +## Inbox retention cleanup + +`inbox-cleaner` removes only Inbox files whose archival was completed by the +handler at least `INBOX_RETENTION_DAYS` ago (90 by default). It confirms that +the accession is still registered in Vault DB and that the corresponding Vault +file exists before removing the Inbox source. + +`INBOX_CLEANUP_INTERVAL_HOURS` controls how often it runs and defaults to `24`. + +The cleaner starts in safe mode. Keep `INBOX_CLEANUP_DRY_RUN=true` until its +logs show the expected candidates and the service has permission to remove a +test Inbox file. Existing Inbox user directories must grant group write access +to `lega`; otherwise the cleaner records the error and does not delete the +file. Set `INBOX_CLEANUP_DRY_RUN=false` only after this check succeeds. The +cleaner records completion timestamp, source path, size and mtime in Vault DB; +files completed before this feature is deployed are intentionally not candidates. + +For an existing Vault DB, apply the migration once after `vault-db` is up: + +```bash +make apply-inbox-cleanup-migration +podman compose up -d inbox-cleaner +``` + +Review its output with `podman logs inbox-cleaner`. A one-off dry run is also +available with `podman compose run --rm --no-deps inbox-cleaner --once`. + Note that the `mq` component will try to create a federated queue to another RabbitMQ server. In `cega` folder, you will find the necessary components to fake Central EGA, and test your local deployment in isolation. diff --git a/deploy/docker/docker-compose.yml b/deploy/docker/docker-compose.yml index 1bda9a35..4b634517 100644 --- a/deploy/docker/docker-compose.yml +++ b/deploy/docker/docker-compose.yml @@ -85,6 +85,34 @@ services: - vault command: "/etc/ega/lega.ini" + inbox-cleaner: + depends_on: + - vault-db + hostname: inbox-cleaner + image: fega/handler:latest + container_name: inbox-cleaner + group_add: + - "${INBOX_GID:-1000}" + env_file: + - .env + environment: + LEGA_LOG: ${LEGA_LOG:-debug} + LEGA_DB_CONNECTION: postgres://lega:${LEGA_DB_PASSWORD:?set LEGA_DB_PASSWORD in deploy/docker/.env}@vault-db:5432/ega?application_name=LocalEGAInboxCleaner + INBOX_RETENTION_DAYS: ${INBOX_RETENTION_DAYS:-90} + INBOX_CLEANUP_DRY_RUN: ${INBOX_CLEANUP_DRY_RUN:-true} + INBOX_CLEANUP_INTERVAL_HOURS: ${INBOX_CLEANUP_INTERVAL_HOURS:-24} + INBOX_CLEANUP_LOG_FILE: /var/log/local/localega/app/cleanup.log + volumes: + - ./lega.ini:/etc/ega/lega.ini:ro + - ${LOCALEGA_DATA_BASE:-/impact_data/lega_data/lega}/inbox:/ega/inbox + - ${LOCALEGA_DATA_BASE:-/impact_data/lega_data/lega}/vault:/ega/vault:ro + - ${LOCALEGA_LOG_DIR:-/var/log/local/localega/app}/inbox-cleaner:/var/log/local/localega/app + networks: + - vault + entrypoint: ["python"] + command: ["-m", "code.cleanup", "/etc/ega/lega.ini"] + restart: unless-stopped + vault-db: hostname: vault image: localega/vault-db:latest diff --git a/src/handler/code/cleanup.py b/src/handler/code/cleanup.py new file mode 100644 index 00000000..0fc32a3d --- /dev/null +++ b/src/handler/code/cleanup.py @@ -0,0 +1,169 @@ +"""Periodic cleanup of Inbox files whose archival has completed.""" + +import argparse +import asyncio +import logging +import os +import sys +from pathlib import Path, PurePosixPath + +from .utils import conf + +LOG = logging.getLogger(__name__) + +CANDIDATES_QUERY = """ +SELECT c.id, c.username, c.filepath, c.accession_id, c.vault_relative_path, + c.inbox_filesize, c.inbox_mtime_ns, c.completed_at, + f.relative_path AS registered_vault_relative_path +FROM private.inbox_cleanup_table AS c +JOIN private.file_table AS f ON f.stable_id = c.accession_id +WHERE c.deleted_at IS NULL + AND c.completed_at <= now() - ($1::bigint * interval '1 day') +ORDER BY c.completed_at, c.id +""" + +MARK_DELETED_QUERY = """ +UPDATE private.inbox_cleanup_table +SET deleted_at = now(), delete_error = NULL +WHERE id = $1 AND deleted_at IS NULL +""" + +MARK_ERROR_QUERY = """ +UPDATE private.inbox_cleanup_table +SET delete_error = $2 +WHERE id = $1 AND deleted_at IS NULL +""" + + +def env_bool(name, default): + value = os.getenv(name) + if value is None: + return default + return value.strip().lower() in {'1', 'true', 'yes', 'on'} + + +def env_positive_int(name, default): + value = os.getenv(name, str(default)) + try: + parsed = int(value) + except ValueError as error: + raise ValueError(f'{name} must be an integer') from error + if parsed < 1: + raise ValueError(f'{name} must be greater than zero') + return parsed + + +def setup_persistent_log(): + """Keep an audit trail even when the container runtime rotates logs.""" + path = Path(os.getenv( + 'INBOX_CLEANUP_LOG_FILE', + '/var/log/local/localega/app/cleanup.log', + )) + try: + path.parent.mkdir(parents=True, exist_ok=True) + handler = logging.FileHandler(path) + handler.setFormatter(logging.Formatter( + '%(asctime)s %(levelname)s %(name)s %(message)s', + )) + LOG.addHandler(handler) + except OSError as error: + LOG.warning('Persistent cleanup log is unavailable at %s: %s', path, error) + + +def safe_relative_path(value): + path = PurePosixPath(value.lstrip('/')) + if not value or path.is_absolute() or '..' in path.parts: + raise ValueError(f'unsafe relative path: {value!r}') + return path + + +def safe_username(value): + if not value or '/' in value or value in {'.', '..'}: + raise ValueError(f'unsafe Inbox username: {value!r}') + return value + + +def path_below(root, relative_path): + root = root.resolve() + path = (root / relative_path).resolve() + if root not in path.parents: + raise ValueError(f'path escapes root: {relative_path}') + return path + + +async def run_once(config, retention_days, dry_run): + connection = config.db + if not connection.connection: + await connection.connect() + + rows = await connection.connection.fetch(CANDIDATES_QUERY, retention_days) + inbox_root = Path(config.get('inbox', 'location', raw=True) % '') + vault_root = Path(config.get('vault', 'location', raw=True)) + summary = {'candidates': len(rows), 'deleted': 0, 'skipped': 0, 'errors': 0} + + for row in rows: + try: + inbox_path = path_below( + inbox_root / safe_username(row['username']), + safe_relative_path(row['filepath']), + ) + vault_relative_path = safe_relative_path(row['vault_relative_path']) + vault_path = path_below(vault_root, vault_relative_path) + + if row['vault_relative_path'] != row['registered_vault_relative_path']: + raise ValueError('vault path no longer matches the registered accession') + if not vault_path.is_file(): + raise FileNotFoundError(f'vault file missing: {vault_path}') + if not inbox_path.is_file(): + LOG.info('Skipping already absent Inbox file for %s: %s', row['accession_id'], inbox_path) + await connection.connection.execute(MARK_DELETED_QUERY, row['id']) + summary['skipped'] += 1 + continue + + source_stat = inbox_path.stat() + if (source_stat.st_size != row['inbox_filesize'] or + source_stat.st_mtime_ns != row['inbox_mtime_ns']): + raise ValueError('Inbox file no longer matches the archived source') + + if dry_run: + LOG.info('DRY RUN: would delete Inbox file for %s: %s', row['accession_id'], inbox_path) + summary['skipped'] += 1 + continue + + inbox_path.unlink() + await connection.connection.execute(MARK_DELETED_QUERY, row['id']) + LOG.info('Deleted Inbox file for %s: %s', row['accession_id'], inbox_path) + summary['deleted'] += 1 + except Exception as error: + LOG.error('Could not clean Inbox record %s: %s', row['id'], error) + await connection.connection.execute(MARK_ERROR_QUERY, row['id'], str(error)) + summary['errors'] += 1 + + LOG.info('Inbox cleanup summary: %s', summary) + return summary + + +async def main(conf_file, once): + config = conf.Configuration(conf_file) + setup_persistent_log() + retention_days = env_positive_int('INBOX_RETENTION_DAYS', 90) + interval_hours = env_positive_int('INBOX_CLEANUP_INTERVAL_HOURS', 24) + dry_run = env_bool('INBOX_CLEANUP_DRY_RUN', True) + + while True: + await run_once(config, retention_days, dry_run) + if once: + return + await asyncio.sleep(interval_hours * 3600) + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('conf_file', help='LocalEGA handler configuration file') + parser.add_argument('--once', action='store_true', help='run one cleanup cycle and exit') + args = parser.parse_args() + try: + asyncio.run(main(args.conf_file, args.once)) + except Exception as error: + LOG.error('Inbox cleanup failed: %r', error, exc_info=True) + sys.exit(2) diff --git a/src/handler/code/handlers/accession.py b/src/handler/code/handlers/accession.py index 8f154449..2b615aab 100644 --- a/src/handler/code/handlers/accession.py +++ b/src/handler/code/handlers/accession.py @@ -53,6 +53,25 @@ async def send_completion(config, staging_path, message): await config.mq.cega_publish(message.parsed, 'files.completed', correlation_id=message.header.properties.content_type) +async def record_inbox_cleanup(config, inbox_path, username, filepath, accession_id, relative_path): + """Record a completed archival for the asynchronous Inbox cleaner.""" + source = Path(inbox_path) + try: + source_stat = source.stat() + except FileNotFoundError: + LOG.warning('Inbox source disappeared before cleanup could be recorded: %s', source) + return + + await config.db.record_inbox_cleanup( + username, + filepath, + accession_id, + relative_path, + source_stat.st_size, + source_stat.st_mtime_ns, + ) + + async def execute(config, message): data = message.parsed @@ -62,11 +81,13 @@ async def execute(config, message): LOG.info('Processing %s: %s', username, filepath) + inbox_prefix = config.get('inbox', 'location', raw=True) staging_prefix = config.get('staging', 'location', raw=True) vault_prefix = config.get('vault', 'location') backup_prefix = config.get('backup', 'location', fallback=None) backup_enabled = bool(backup_prefix) + inbox_path = os.path.join(inbox_prefix % username, filepath.strip('/') ) staging_path = os.path.join(staging_prefix % username, filepath.strip('/') ) LOG.debug('Staging path: %s', staging_path) @@ -86,6 +107,14 @@ async def execute(config, message): if vpath.exists(): # do nothing and return early LOG.info('Vault path already exists') + await record_inbox_cleanup( + config, + inbox_path, + username, + filepath, + accession_id, + relative_path, + ) return await send_completion(config, staging_path, message) vpath.parent.mkdir(parents=True, exist_ok=True) @@ -180,6 +209,14 @@ async def execute(config, message): accession_id, relative_path) + await record_inbox_cleanup( + config, + inbox_path, + username, + filepath, + accession_id, + relative_path, + ) + # All good: send completion return await send_completion(config, staging_path, message) - diff --git a/src/handler/code/utils/db.py b/src/handler/code/utils/db.py index c04a9925..cbccfb6f 100644 --- a/src/handler/code/utils/db.py +++ b/src/handler/code/utils/db.py @@ -123,6 +123,16 @@ async def save_file(self, *args, **kwargs): return await self.connection.fetchval(query, *args, **kwargs) + async def record_inbox_cleanup(self, *args): + """Persist the source details needed by the asynchronous Inbox cleaner.""" + if not self.connection: + await self.connect() + + return await self.connection.fetchval( + 'SELECT private.record_inbox_cleanup($1, $2, $3, $4, $5, $6)', + *args, + ) + async def fetchval(self, stmt, *args, **kwargs): query = self.conf.get(self.conf_section, stmt, raw=True) diff --git a/src/vault/initdb.d/35-inbox-cleanup.sql b/src/vault/initdb.d/35-inbox-cleanup.sql new file mode 100644 index 00000000..07e54ada --- /dev/null +++ b/src/vault/initdb.d/35-inbox-cleanup.sql @@ -0,0 +1,55 @@ +-- Inbox retention records. These rows are created only after a file has +-- been written, verified and registered in the vault. +CREATE TABLE IF NOT EXISTS private.inbox_cleanup_table ( + id BIGSERIAL PRIMARY KEY, + username text NOT NULL, + filepath text NOT NULL, + accession_id text NOT NULL REFERENCES public.file_table(stable_id), + vault_relative_path text NOT NULL, + inbox_filesize bigint NOT NULL, + inbox_mtime_ns bigint NOT NULL, + completed_at timestamp(6) with time zone NOT NULL DEFAULT now(), + deleted_at timestamp(6) with time zone, + delete_error text, + + UNIQUE (username, filepath, accession_id, inbox_mtime_ns) +); + +CREATE INDEX IF NOT EXISTS inbox_cleanup_pending_idx + ON private.inbox_cleanup_table (completed_at) + WHERE deleted_at IS NULL; + +CREATE OR REPLACE FUNCTION private.record_inbox_cleanup( + _username text, + _filepath text, + _accession_id text, + _vault_relative_path text, + _inbox_filesize bigint, + _inbox_mtime_ns bigint +) +RETURNS void +LANGUAGE sql +AS $_$ + INSERT INTO private.inbox_cleanup_table ( + username, + filepath, + accession_id, + vault_relative_path, + inbox_filesize, + inbox_mtime_ns + ) VALUES ( + _username, + _filepath, + _accession_id, + _vault_relative_path, + _inbox_filesize, + _inbox_mtime_ns + ) + ON CONFLICT (username, filepath, accession_id, inbox_mtime_ns) DO NOTHING; +$_$; + +-- Keep migration self-contained for already initialised Vault databases. +GRANT USAGE ON SCHEMA private TO lega; +GRANT USAGE ON SEQUENCE private.inbox_cleanup_table_id_seq TO lega; +GRANT SELECT,INSERT,UPDATE ON TABLE private.inbox_cleanup_table TO lega; +GRANT EXECUTE ON FUNCTION private.record_inbox_cleanup(text, text, text, text, bigint, bigint) TO lega; diff --git a/src/vault/initdb.d/90-grants.sql b/src/vault/initdb.d/90-grants.sql index 194a7e0f..cfbe833d 100644 --- a/src/vault/initdb.d/90-grants.sql +++ b/src/vault/initdb.d/90-grants.sql @@ -14,12 +14,15 @@ GRANT SELECT,INSERT,DELETE ON TABLE public.dataset_file_table TO lega; GRANT USAGE ON SCHEMA private TO lega; GRANT SELECT,INSERT,UPDATE,DELETE ON TABLE private.file_table TO lega; +GRANT USAGE ON SEQUENCE private.inbox_cleanup_table_id_seq TO lega; +GRANT SELECT,INSERT,UPDATE ON TABLE private.inbox_cleanup_table TO lega; GRANT USAGE ON SEQUENCE private.dataset_permission_table_id_seq TO lega; GRANT SELECT,INSERT,UPDATE,DELETE ON TABLE private.dataset_permission_table TO lega; GRANT SELECT,INSERT,UPDATE,DELETE ON TABLE private.user_password_table TO lega; GRANT EXECUTE ON FUNCTION public.extract_name(text) TO lega; GRANT EXECUTE ON FUNCTION public.upsert_file TO lega; +GRANT EXECUTE ON FUNCTION private.record_inbox_cleanup(text, text, text, text, bigint, bigint) TO lega; GRANT EXECUTE ON FUNCTION public.process_dac_dataset_message(jsonb) TO lega; GRANT EXECUTE ON FUNCTION public.process_mapping_message(jsonb) TO lega; GRANT EXECUTE ON FUNCTION public.process_release_message TO lega; @@ -32,4 +35,3 @@ GRANT EXECUTE ON FUNCTION public.process_user_contact_message TO le GRANT USAGE ON SCHEMA crypt4gh TO lega; GRANT EXECUTE ON FUNCTION crypt4gh.parse_pubkey TO lega; - From 33a99f139963a3fce00c37fd270c85a26c6f96e0 Mon Sep 17 00:00:00 2001 From: Alejandro Date: Wed, 22 Jul 2026 12:15:39 +0200 Subject: [PATCH 2/3] Harden Inbox retention cleanup --- deploy/docker/.env.example | 1 + deploy/docker/README.md | 5 + deploy/docker/docker-compose.yml | 1 + src/handler/Makefile | 5 +- src/handler/code/cleanup.py | 83 +++++++++--- src/handler/code/handlers/accession.py | 25 ++-- src/handler/tests/test_cleanup.py | 179 +++++++++++++++++++++++++ 7 files changed, 271 insertions(+), 28 deletions(-) create mode 100644 src/handler/tests/test_cleanup.py diff --git a/deploy/docker/.env.example b/deploy/docker/.env.example index 87eadfc1..ac1cec78 100644 --- a/deploy/docker/.env.example +++ b/deploy/docker/.env.example @@ -51,3 +51,4 @@ LEGA_LOG=debug INBOX_RETENTION_DAYS=90 INBOX_CLEANUP_DRY_RUN=true INBOX_CLEANUP_INTERVAL_HOURS=24 +INBOX_CLEANUP_BATCH_SIZE=1000 diff --git a/deploy/docker/README.md b/deploy/docker/README.md index 49dd8d2c..11c2f3e5 100644 --- a/deploy/docker/README.md +++ b/deploy/docker/README.md @@ -142,6 +142,8 @@ the accession is still registered in Vault DB and that the corresponding Vault file exists before removing the Inbox source. `INBOX_CLEANUP_INTERVAL_HOURS` controls how often it runs and defaults to `24`. +`INBOX_CLEANUP_BATCH_SIZE` limits each cycle to `1000` candidates by default, +preventing an accumulated backlog from causing an unbounded cleanup run. The cleaner starts in safe mode. Keep `INBOX_CLEANUP_DRY_RUN=true` until its logs show the expected candidates and the service has permission to remove a @@ -150,6 +152,9 @@ to `lega`; otherwise the cleaner records the error and does not delete the file. Set `INBOX_CLEANUP_DRY_RUN=false` only after this check succeeds. The cleaner records completion timestamp, source path, size and mtime in Vault DB; files completed before this feature is deployed are intentionally not candidates. +Invalid boolean values are rejected, so a typo cannot accidentally disable dry-run +mode. Symlinks are never followed, and the registered Vault payload size is checked +before an Inbox file is removed. For an existing Vault DB, apply the migration once after `vault-db` is up: diff --git a/deploy/docker/docker-compose.yml b/deploy/docker/docker-compose.yml index 4b634517..0b4cbc72 100644 --- a/deploy/docker/docker-compose.yml +++ b/deploy/docker/docker-compose.yml @@ -101,6 +101,7 @@ services: INBOX_RETENTION_DAYS: ${INBOX_RETENTION_DAYS:-90} INBOX_CLEANUP_DRY_RUN: ${INBOX_CLEANUP_DRY_RUN:-true} INBOX_CLEANUP_INTERVAL_HOURS: ${INBOX_CLEANUP_INTERVAL_HOURS:-24} + INBOX_CLEANUP_BATCH_SIZE: ${INBOX_CLEANUP_BATCH_SIZE:-1000} INBOX_CLEANUP_LOG_FILE: /var/log/local/localega/app/cleanup.log volumes: - ./lega.ini:/etc/ega/lega.ini:ro diff --git a/src/handler/Makefile b/src/handler/Makefile index c4e60cee..f922f0fb 100644 --- a/src/handler/Makefile +++ b/src/handler/Makefile @@ -4,10 +4,13 @@ ARGS = IMG=fega/handler:$(COMMIT) -.PHONY: latest build +.PHONY: latest build test all: latest +test: + PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=. python3 -m unittest discover -s tests -v + ARCH=$(shell uname -m) ifeq ($(ARCH), arm64) # reset for MacOS ARCH=aarch64 diff --git a/src/handler/code/cleanup.py b/src/handler/code/cleanup.py index 0fc32a3d..08813513 100644 --- a/src/handler/code/cleanup.py +++ b/src/handler/code/cleanup.py @@ -4,22 +4,25 @@ import asyncio import logging import os +import stat import sys from pathlib import Path, PurePosixPath -from .utils import conf - LOG = logging.getLogger(__name__) CANDIDATES_QUERY = """ SELECT c.id, c.username, c.filepath, c.accession_id, c.vault_relative_path, c.inbox_filesize, c.inbox_mtime_ns, c.completed_at, - f.relative_path AS registered_vault_relative_path + f.relative_path AS registered_vault_relative_path, + f.payload_size AS registered_vault_filesize FROM private.inbox_cleanup_table AS c JOIN private.file_table AS f ON f.stable_id = c.accession_id WHERE c.deleted_at IS NULL AND c.completed_at <= now() - ($1::bigint * interval '1 day') -ORDER BY c.completed_at, c.id +-- Process never-attempted rows first so persistent errors cannot starve newer +-- valid candidates when a batch limit is in use. +ORDER BY (c.delete_error IS NOT NULL), c.completed_at, c.id +LIMIT $2::bigint """ MARK_DELETED_QUERY = """ @@ -39,7 +42,14 @@ def env_bool(name, default): value = os.getenv(name) if value is None: return default - return value.strip().lower() in {'1', 'true', 'yes', 'on'} + value = value.strip().lower() + if value in {'1', 'true', 'yes', 'on'}: + return True + if value in {'0', 'false', 'no', 'off'}: + return False + raise ValueError( + f'{name} must be one of true/false, yes/no, on/off or 1/0' + ) def env_positive_int(name, default): @@ -72,7 +82,8 @@ def setup_persistent_log(): def safe_relative_path(value): path = PurePosixPath(value.lstrip('/')) - if not value or path.is_absolute() or '..' in path.parts: + if (not value or path == PurePosixPath('.') or path.is_absolute() or + '..' in path.parts): raise ValueError(f'unsafe relative path: {value!r}') return path @@ -85,42 +96,77 @@ def safe_username(value): def path_below(root, relative_path): root = root.resolve() - path = (root / relative_path).resolve() - if root not in path.parents: + path = root.joinpath(*relative_path.parts) + try: + path.relative_to(root) + except ValueError as error: + raise ValueError(f'path escapes root: {relative_path}') from error + if path == root: raise ValueError(f'path escapes root: {relative_path}') return path -async def run_once(config, retention_days, dry_run): +def regular_file_stat(root, path): + """Return lstat data while rejecting symlinks in every path component.""" + root = root.resolve() + relative_path = path.relative_to(root) + current = root + parts = relative_path.parts + + for index, part in enumerate(parts): + current = current / part + current_stat = current.lstat() + if stat.S_ISLNK(current_stat.st_mode): + raise ValueError(f'symbolic links are not allowed: {current}') + if index < len(parts) - 1 and not stat.S_ISDIR(current_stat.st_mode): + raise ValueError(f'path component is not a directory: {current}') + + if not stat.S_ISREG(current_stat.st_mode): + raise ValueError(f'path is not a regular file: {path}') + return current_stat + + +async def run_once(config, retention_days, dry_run, batch_size=1000): connection = config.db if not connection.connection: await connection.connect() - rows = await connection.connection.fetch(CANDIDATES_QUERY, retention_days) + rows = await connection.connection.fetch( + CANDIDATES_QUERY, + retention_days, + batch_size, + ) inbox_root = Path(config.get('inbox', 'location', raw=True) % '') vault_root = Path(config.get('vault', 'location', raw=True)) summary = {'candidates': len(rows), 'deleted': 0, 'skipped': 0, 'errors': 0} for row in rows: try: + username = safe_username(row['username']) inbox_path = path_below( - inbox_root / safe_username(row['username']), - safe_relative_path(row['filepath']), + inbox_root, + PurePosixPath(username) / safe_relative_path(row['filepath']), ) vault_relative_path = safe_relative_path(row['vault_relative_path']) vault_path = path_below(vault_root, vault_relative_path) if row['vault_relative_path'] != row['registered_vault_relative_path']: raise ValueError('vault path no longer matches the registered accession') - if not vault_path.is_file(): - raise FileNotFoundError(f'vault file missing: {vault_path}') - if not inbox_path.is_file(): + vault_stat = regular_file_stat(vault_root, vault_path) + if vault_stat.st_size != row['registered_vault_filesize']: + raise ValueError('Vault file size no longer matches the registered accession') + + try: + source_stat = regular_file_stat( + inbox_root, + inbox_path, + ) + except FileNotFoundError: LOG.info('Skipping already absent Inbox file for %s: %s', row['accession_id'], inbox_path) await connection.connection.execute(MARK_DELETED_QUERY, row['id']) summary['skipped'] += 1 continue - source_stat = inbox_path.stat() if (source_stat.st_size != row['inbox_filesize'] or source_stat.st_mtime_ns != row['inbox_mtime_ns']): raise ValueError('Inbox file no longer matches the archived source') @@ -144,14 +190,17 @@ async def run_once(config, retention_days, dry_run): async def main(conf_file, once): + from .utils import conf + config = conf.Configuration(conf_file) setup_persistent_log() retention_days = env_positive_int('INBOX_RETENTION_DAYS', 90) interval_hours = env_positive_int('INBOX_CLEANUP_INTERVAL_HOURS', 24) + batch_size = env_positive_int('INBOX_CLEANUP_BATCH_SIZE', 1000) dry_run = env_bool('INBOX_CLEANUP_DRY_RUN', True) while True: - await run_once(config, retention_days, dry_run) + await run_once(config, retention_days, dry_run, batch_size) if once: return await asyncio.sleep(interval_hours * 3600) diff --git a/src/handler/code/handlers/accession.py b/src/handler/code/handlers/accession.py index 2b615aab..4b2bcd0c 100644 --- a/src/handler/code/handlers/accession.py +++ b/src/handler/code/handlers/accession.py @@ -58,18 +58,23 @@ async def record_inbox_cleanup(config, inbox_path, username, filepath, accession source = Path(inbox_path) try: source_stat = source.stat() + await config.db.record_inbox_cleanup( + username, + filepath, + accession_id, + relative_path, + source_stat.st_size, + source_stat.st_mtime_ns, + ) except FileNotFoundError: LOG.warning('Inbox source disappeared before cleanup could be recorded: %s', source) - return - - await config.db.record_inbox_cleanup( - username, - filepath, - accession_id, - relative_path, - source_stat.st_size, - source_stat.st_mtime_ns, - ) + except Exception: + # Cleanup is best-effort: archival completion must not fail merely + # because the optional retention receipt could not be persisted. + LOG.exception( + 'Could not record Inbox cleanup receipt for %s; source retained', + accession_id, + ) async def execute(config, message): diff --git a/src/handler/tests/test_cleanup.py b/src/handler/tests/test_cleanup.py new file mode 100644 index 00000000..7b4f68dd --- /dev/null +++ b/src/handler/tests/test_cleanup.py @@ -0,0 +1,179 @@ +import os +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from code import cleanup + + +class FakePool: + def __init__(self, rows): + self.rows = rows + self.fetch_args = None + self.executed = [] + + async def fetch(self, query, *args): + self.fetch_args = args + return self.rows + + async def execute(self, query, *args): + self.executed.append((query, args)) + + +class FakeDB: + def __init__(self, rows): + self.connection = FakePool(rows) + + +class FakeConfig: + def __init__(self, inbox_root, vault_root, rows): + self.db = FakeDB(rows) + self.inbox_root = inbox_root + self.vault_root = vault_root + + def get(self, section, option, raw=False): + if section == 'inbox': + return str(self.inbox_root / '%s') + if section == 'vault': + return str(self.vault_root) + raise KeyError((section, option)) + + +class CleanupValidationTests(unittest.TestCase): + def test_env_bool_rejects_invalid_value(self): + with mock.patch.dict(os.environ, {'CLEANUP_BOOL': 'treu'}): + with self.assertRaisesRegex(ValueError, 'CLEANUP_BOOL'): + cleanup.env_bool('CLEANUP_BOOL', True) + + def test_safe_relative_path_rejects_root_and_traversal(self): + for value in ('', '/', '.', '../file', 'dir/../../file'): + with self.subTest(value=value): + with self.assertRaises(ValueError): + cleanup.safe_relative_path(value) + + +class CleanupRunTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.temporary_directory = tempfile.TemporaryDirectory() + root = Path(self.temporary_directory.name) + self.inbox_root = root / 'inbox' + self.vault_root = root / 'vault' + self.user_root = self.inbox_root / 'user@example.org' + self.user_root.mkdir(parents=True) + self.vault_root.mkdir() + + def tearDown(self): + self.temporary_directory.cleanup() + + def make_row(self, source_path, vault_path): + source_stat = source_path.stat() + vault_stat = vault_path.stat() + return { + 'id': 7, + 'username': 'user@example.org', + 'filepath': '/' + source_path.name, + 'accession_id': 'EGAF000000001', + 'vault_relative_path': vault_path.name, + 'inbox_filesize': source_stat.st_size, + 'inbox_mtime_ns': source_stat.st_mtime_ns, + 'completed_at': None, + 'registered_vault_relative_path': vault_path.name, + 'registered_vault_filesize': vault_stat.st_size, + } + + async def test_deletes_only_matching_regular_file(self): + source_path = self.user_root / 'sample.c4gh' + vault_path = self.vault_root / 'EGAF000000001' + source_path.write_bytes(b'inbox payload') + vault_path.write_bytes(b'vault payload') + config = FakeConfig( + self.inbox_root, + self.vault_root, + [self.make_row(source_path, vault_path)], + ) + + summary = await cleanup.run_once(config, 90, False, batch_size=25) + + self.assertFalse(source_path.exists()) + self.assertEqual(summary['deleted'], 1) + self.assertEqual(config.db.connection.fetch_args, (90, 25)) + self.assertEqual(config.db.connection.executed[0][1], (7,)) + + async def test_dry_run_keeps_file(self): + source_path = self.user_root / 'sample.c4gh' + vault_path = self.vault_root / 'EGAF000000001' + source_path.write_bytes(b'inbox payload') + vault_path.write_bytes(b'vault payload') + config = FakeConfig( + self.inbox_root, + self.vault_root, + [self.make_row(source_path, vault_path)], + ) + + summary = await cleanup.run_once(config, 90, True) + + self.assertTrue(source_path.exists()) + self.assertEqual(summary['skipped'], 1) + self.assertEqual(config.db.connection.executed, []) + + async def test_rejects_symlink_without_deleting_target(self): + target_path = self.user_root / 'other.c4gh' + source_path = self.user_root / 'sample.c4gh' + vault_path = self.vault_root / 'EGAF000000001' + target_path.write_bytes(b'inbox payload') + source_path.symlink_to(target_path) + vault_path.write_bytes(b'vault payload') + config = FakeConfig( + self.inbox_root, + self.vault_root, + [self.make_row(source_path, vault_path)], + ) + + summary = await cleanup.run_once(config, 90, False) + + self.assertTrue(source_path.is_symlink()) + self.assertTrue(target_path.exists()) + self.assertEqual(summary['errors'], 1) + self.assertIn('symbolic links', config.db.connection.executed[0][1][1]) + + async def test_rejects_symlinked_user_directory(self): + external_root = self.inbox_root.parent / 'external-user-directory' + external_root.mkdir() + source_path = external_root / 'sample.c4gh' + vault_path = self.vault_root / 'EGAF000000001' + source_path.write_bytes(b'inbox payload') + vault_path.write_bytes(b'vault payload') + self.user_root.rmdir() + self.user_root.symlink_to(external_root, target_is_directory=True) + config = FakeConfig( + self.inbox_root, + self.vault_root, + [self.make_row(source_path, vault_path)], + ) + config.db.connection.rows[0]['filepath'] = '/sample.c4gh' + + summary = await cleanup.run_once(config, 90, False) + + self.assertTrue(source_path.exists()) + self.assertEqual(summary['errors'], 1) + self.assertIn('symbolic links', config.db.connection.executed[0][1][1]) + + async def test_rejects_vault_size_mismatch(self): + source_path = self.user_root / 'sample.c4gh' + vault_path = self.vault_root / 'EGAF000000001' + source_path.write_bytes(b'inbox payload') + vault_path.write_bytes(b'vault payload') + row = self.make_row(source_path, vault_path) + row['registered_vault_filesize'] += 1 + config = FakeConfig(self.inbox_root, self.vault_root, [row]) + + summary = await cleanup.run_once(config, 90, False) + + self.assertTrue(source_path.exists()) + self.assertEqual(summary['errors'], 1) + self.assertIn('Vault file size', config.db.connection.executed[0][1][1]) + + +if __name__ == '__main__': + unittest.main() From 72dbe7e47183cd22be2e42961682d1016624c8e4 Mon Sep 17 00:00:00 2001 From: Alejandro Date: Wed, 22 Jul 2026 12:30:43 +0200 Subject: [PATCH 3/3] Fix one-off Inbox cleaner invocation --- deploy/docker/docker-compose.yml | 4 ++-- src/handler/code/cleanup.py | 7 ++++++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/deploy/docker/docker-compose.yml b/deploy/docker/docker-compose.yml index 0b4cbc72..4c83f4d8 100644 --- a/deploy/docker/docker-compose.yml +++ b/deploy/docker/docker-compose.yml @@ -110,8 +110,8 @@ services: - ${LOCALEGA_LOG_DIR:-/var/log/local/localega/app}/inbox-cleaner:/var/log/local/localega/app networks: - vault - entrypoint: ["python"] - command: ["-m", "code.cleanup", "/etc/ega/lega.ini"] + entrypoint: ["python", "-m", "code.cleanup"] + command: ["/etc/ega/lega.ini"] restart: unless-stopped vault-db: diff --git a/src/handler/code/cleanup.py b/src/handler/code/cleanup.py index 08813513..5c54bbb8 100644 --- a/src/handler/code/cleanup.py +++ b/src/handler/code/cleanup.py @@ -208,7 +208,12 @@ async def main(conf_file, once): if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument('conf_file', help='LocalEGA handler configuration file') + parser.add_argument( + 'conf_file', + nargs='?', + default='/etc/ega/lega.ini', + help='LocalEGA handler configuration file (default: /etc/ega/lega.ini)', + ) parser.add_argument('--once', action='store_true', help='run one cleanup cycle and exit') args = parser.parse_args() try: