diff --git a/config/vector.yaml b/config/vector.yaml index 2e29a157a..d458acdfb 100644 --- a/config/vector.yaml +++ b/config/vector.yaml @@ -75,22 +75,22 @@ transforms: # Sinks - write to files sinks: - # Platform services log file + # Platform services log file (daily rotation) platform_logs: type: file inputs: - route_platform - path: /data/logs/platform.json + path: /data/logs/platform-%Y-%m-%d.json encoding: codec: json compression: none - # Agent containers log file + # Agent containers log file (daily rotation) agent_logs: type: file inputs: - route_agents - path: /data/logs/agents.json + path: /data/logs/agents-%Y-%m-%d.json encoding: codec: json compression: none diff --git a/docker-compose.yml b/docker-compose.yml index 2baf00003..bd167246b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -31,6 +31,11 @@ services: - EMAIL_PROVIDER=${EMAIL_PROVIDER:-resend} - RESEND_API_KEY=${RESEND_API_KEY:-} - SMTP_FROM=${SMTP_FROM:-noreply@trinity.example.com} + # Log Retention & Archival Configuration + - LOG_RETENTION_DAYS=${LOG_RETENTION_DAYS:-90} + - LOG_ARCHIVE_ENABLED=${LOG_ARCHIVE_ENABLED:-true} + - LOG_CLEANUP_HOUR=${LOG_CLEANUP_HOUR:-3} + - LOG_ARCHIVE_PATH=/data/archives # Local path for archived logs volumes: - /var/run/docker.sock:/var/run/docker.sock:ro - ./src/backend:/app @@ -38,6 +43,8 @@ services: - ./config/trinity-meta-prompt:/trinity-meta-prompt:ro - agent-configs:/agent-configs - trinity-data:/data + - trinity-logs:/data/logs:ro # Read-only access to Vector logs + - trinity-archives:/data/archives # Log archives storage depends_on: redis: condition: service_started @@ -181,6 +188,7 @@ volumes: redis-data: trinity-data: trinity-logs: # Vector log storage + trinity-archives: # Compressed log archives networks: trinity-network: diff --git a/docker/backend/Dockerfile b/docker/backend/Dockerfile index b26114fcb..39db77f84 100644 --- a/docker/backend/Dockerfile +++ b/docker/backend/Dockerfile @@ -32,6 +32,7 @@ COPY ../../src/backend/db_models.py /app/ COPY ../../src/backend/dependencies.py /app/ COPY ../../src/backend/credentials.py /app/ COPY ../../src/backend/database.py /app/ +COPY ../../src/backend/logging_config.py /app/ # Copy module directories COPY ../../src/backend/routers /app/routers/ diff --git a/docs/memory/changelog.md b/docs/memory/changelog.md index e5398d391..a0a85c0fe 100644 --- a/docs/memory/changelog.md +++ b/docs/memory/changelog.md @@ -1,3 +1,105 @@ +### 2026-01-06 18:30:00 +♻️ **Refactor: Sovereign Archive Storage Architecture** + +**Refactoring**: Removed external S3 dependency and implemented pluggable storage interface with local-only default for sovereign deployments. + +**Motivation**: Following architectural feedback to maintain full data sovereignty - no system logs or archives should leave the company's infrastructure. The platform now stores all data locally within mounted volumes, giving operators full control over backup strategies. + +**What Changed**: +1. **New Storage Interface** - Abstract `ArchiveStorage` base class with pluggable backend support +2. **Local Storage Default** - `LocalArchiveStorage` implementation stores archives in mounted Docker volume +3. **S3 Code Removed** - Deleted S3-specific code (`s3_storage.py`), boto3 dependency, and S3 env vars +4. **Sovereign Backup Strategies** - Documentation now covers Docker volume backups, NAS/NFS mounts, rsync, cross-instance copies + +**New Files**: +- `src/backend/services/archive_storage.py` - Storage abstraction layer + +**Deleted Files**: +- `src/backend/services/s3_storage.py` - S3 integration (removed) + +**Modified Files**: +- `src/backend/services/log_archive_service.py` - Uses storage interface instead of S3 directly +- `docker/backend/Dockerfile` - Removed boto3 dependency +- `docker-compose.yml` - Removed S3 env vars, added `LOG_ARCHIVE_PATH` +- `docs/memory/feature-flows/vector-logging.md` - Replaced S3 docs with sovereign backup strategies +- `tests/test_log_archive.py` - Removed S3 tests, updated to use storage interface + +**Configuration Changes**: +```bash +# Removed: +LOG_S3_ENABLED, LOG_S3_BUCKET, LOG_S3_ACCESS_KEY, LOG_S3_SECRET_KEY, +LOG_S3_ENDPOINT, LOG_S3_REGION + +# Added: +LOG_ARCHIVE_PATH=/data/archives # Local path for archived logs +``` + +**Backup Strategies** (Sovereign): +- Mount NAS/NFS to archives volume +- rsync to backup server via cron +- Docker volume backup/restore +- Cross-instance volume copy + +**Impact**: +- **Breaking**: S3 configuration no longer supported (by design) +- **Non-Breaking**: Archives still stored in `trinity-archives` volume +- **Benefit**: Full data sovereignty - no external dependencies +- **Extensible**: Storage interface allows future backends (NFS, SFTP, etc.) without S3 + +**Architecture Philosophy**: Keep all data (logs, archives, backups) within operator-controlled infrastructure. External storage can be implemented via volume mounts or custom scripts, not hardcoded cloud dependencies. + +--- + +### 2026-01-05 18:30:00 +✨ **Feature: Vector Log Retention and Archival** + +**New Feature**: Automated log retention, rotation, and archival system for Vector logs with local storage. + +**Compliance Gap Addressed**: Vector logging lacked retention policy and archival capabilities required for SOC2/ISO27001 compliance. + +**What Was Added**: +1. **Daily Log Rotation** - Vector now writes to date-stamped files (`platform-2026-01-05.json`) +2. **Automated Archival** - Nightly job compresses and archives logs older than retention period +3. **Local Storage** - Archives stored in mounted Docker volume (`trinity-archives`) +4. **API Endpoints** - Admin-only endpoints for stats, config, and manual archival +5. **Integrity Verification** - SHA256 checksums ensure archive integrity + +**New Files**: +- `src/backend/services/log_archive_service.py` - Archival logic + APScheduler +- `src/backend/routers/logs.py` - Log management API endpoints +- `tests/test_log_archive.py` - Comprehensive test coverage + +**Modified Files**: +- `config/vector.yaml` - Date-based file paths for daily rotation +- `src/backend/main.py` - Register logs router, start archive scheduler +- `docker-compose.yml` - Add log retention env vars, trinity-archives volume +- `docs/memory/feature-flows/vector-logging.md` - Document retention features + +**Configuration**: +```bash +LOG_RETENTION_DAYS=90 # Days to keep logs +LOG_ARCHIVE_ENABLED=true # Enable automated archival +LOG_CLEANUP_HOUR=3 # Hour (UTC) to run nightly job +LOG_ARCHIVE_PATH=/data/archives # Local path for archived logs +``` + +**API Endpoints**: +- `GET /api/logs/stats` - Log file statistics +- `GET /api/logs/retention` - Retention configuration +- `PUT /api/logs/retention` - Update retention (runtime) +- `POST /api/logs/archive` - Manual archival trigger +- `GET /api/logs/health` - Service health check + +**Impact**: +- No breaking changes to existing Vector functionality +- Existing single-file logs (`platform.json`, `agents.json`) can be manually archived +- New date-based files start after Vector restart +- Disk space remains stable after retention period + +**Compliance**: Addresses SOC2/ISO27001 requirement for log retention policy and secure archival. + +--- + ### 2026-01-05 10:15:00 🐛 **Fix: ReplayTimeline infinite loop causing browser hang** diff --git a/docs/memory/feature-flows/vector-logging.md b/docs/memory/feature-flows/vector-logging.md index 9a0cf779f..e3827af6a 100644 --- a/docs/memory/feature-flows/vector-logging.md +++ b/docs/memory/feature-flows/vector-logging.md @@ -313,12 +313,159 @@ curl http://localhost:8686/health curl http://localhost:8686/api/v1/topology ``` +## Log Retention & Archival + +### Overview + +Automated log retention system that: +1. Rotates logs daily to date-stamped files +2. Archives old logs with compression to local storage +3. Deletes originals after successful archive + +### Configuration + +Environment variables in `docker-compose.yml`: + +| Variable | Default | Description | +|----------|---------|-------------| +| `LOG_RETENTION_DAYS` | `90` | Days to keep logs before archival | +| `LOG_ARCHIVE_ENABLED` | `true` | Enable automated archival | +| `LOG_CLEANUP_HOUR` | `3` | Hour (UTC) to run nightly archival | +| `LOG_ARCHIVE_PATH` | `/data/archives` | Local path for archived logs | + +### Daily Rotation + +Vector writes logs to date-stamped files: +- `platform-2026-01-05.json` (today's platform logs) +- `agents-2026-01-05.json` (today's agent logs) + +Files rotate automatically at midnight UTC. + +### Automated Archival + +The backend runs a nightly job (default: 3 AM UTC) that: + +1. **Finds old files**: Identifies logs older than retention period +2. **Compresses**: Gzip level 9 (~90% size reduction) +3. **Verifies**: SHA256 integrity check +4. **Stores locally**: Moves to mounted archive volume +5. **Cleans up**: Deletes original files after successful archive + +### API Endpoints + +Admin-only endpoints for log management: + +```bash +# Get log statistics +curl -H "Authorization: Bearer $TOKEN" http://localhost:8000/api/logs/stats + +# Get retention configuration +curl -H "Authorization: Bearer $TOKEN" http://localhost:8000/api/logs/retention + +# Update retention (runtime only) +curl -X PUT -H "Authorization: Bearer $TOKEN" \ + -H "Content-Type: application/json" \ + -d '{"retention_days": 30, "archive_enabled": true, "cleanup_hour": 3}' \ + http://localhost:8000/api/logs/retention + +# Manually trigger archival +curl -X POST -H "Authorization: Bearer $TOKEN" \ + -H "Content-Type: application/json" \ + -d '{"retention_days": 90, "delete_after_archive": true}' \ + http://localhost:8000/api/logs/archive + +# Check archival service health +curl -H "Authorization: Bearer $TOKEN" http://localhost:8000/api/logs/health +``` + +### Custom Backup Strategies + +Trinity archives logs locally by default, keeping all data within your infrastructure (sovereign deployment). You can implement custom backup strategies using standard Docker volume management: + +#### Option 1: Mount NAS/NFS for Archives + +Mount a network storage device to the archives volume: + +```yaml +# In docker-compose.yml +services: + backend: + volumes: + - type: bind + source: /mnt/nas/trinity-archives # Your NAS mount point + target: /data/archives +``` + +#### Option 2: rsync to Backup Server + +Create a cron job to sync archives to another server: + +```bash +# /etc/cron.daily/trinity-archive-sync +#!/bin/bash +rsync -avz --delete \ + /var/lib/docker/volumes/trinity-archives/_data/ \ + backup-server:/backups/trinity-logs/ +``` + +#### Option 3: Docker Volume Backup + +Use Docker's built-in volume backup: + +```bash +# Backup archives volume to tarball +docker run --rm \ + -v trinity-archives:/data \ + -v $(pwd):/backup \ + alpine tar czf /backup/trinity-archives-$(date +%Y%m%d).tar.gz /data + +# Restore from tarball +docker run --rm \ + -v trinity-archives:/data \ + -v $(pwd):/backup \ + alpine tar xzf /backup/trinity-archives-20260105.tar.gz -C / +``` + +#### Option 4: Cross-Instance Volume Copy + +Copy archives between Docker instances: + +```bash +# Export from instance A +docker run --rm \ + -v trinity-archives:/source \ + -v $(pwd):/dest \ + alpine sh -c "cd /source && tar czf - ." > trinity-archives.tar.gz + +# Import to instance B (transfer trinity-archives.tar.gz via scp/rsync) +docker run --rm \ + -v trinity-archives:/dest \ + -v $(pwd):/source \ + alpine sh -c "cd /dest && tar xzf /source/trinity-archives.tar.gz" +``` + +### Storage Locations + +| Location | Contents | Purpose | +|----------|----------|---------| +| `/data/logs/` | Active log files | Daily logs (current + retention period) | +| `/data/archives/` | Compressed archives | Local storage of archived logs | +| Docker volume `trinity-archives` | Persistent storage | Mounted at `/data/archives` in backend | + +### Disk Space Management + +Example with 90-day retention: +- Daily log size: ~150 MB +- 90 days uncompressed: ~13.5 GB +- 90 days compressed: ~1.35 GB (10% of original) +- Active logs stay at ~13.5 GB (stable after 90 days) + ## Side Effects -- **File Output**: Logs written to Docker volume `trinity-logs` +- **File Output**: Logs written to Docker volume `trinity-logs` with daily rotation +- **Compressed Archives**: Old logs archived to Docker volume `trinity-archives` - **No Database**: Logs are file-based, not persisted to SQLite - **No WebSocket**: No real-time log streaming to UI (query only) -- **No Rotation**: Currently no automatic rotation configured (monitor file sizes) ## Error Handling @@ -442,4 +589,6 @@ No cleanup required - logs accumulate until manually cleared. | Date | Change | |------|--------| +| 2026-01-06 | Refactored to sovereign architecture - removed external S3 dependency, added pluggable storage interface | +| 2026-01-05 | Added log retention, rotation, and archival capabilities | | 2025-12-31 | Initial implementation - replaced audit-logger with Vector | diff --git a/src/backend/main.py b/src/backend/main.py index c06818e83..eb8c6e6e7 100644 --- a/src/backend/main.py +++ b/src/backend/main.py @@ -45,6 +45,7 @@ from routers.public import router as public_router from routers.setup import router as setup_router from routers.telemetry import router as telemetry_router +from routers.logs import router as logs_router # Import scheduler service from services.scheduler_service import scheduler_service @@ -55,6 +56,9 @@ # Import system agent service from services.system_agent_service import system_agent_service +# Import log archive service +from services.log_archive_service import log_archive_service + # Import credentials manager for GitHub PAT initialization from credentials import CredentialManager, CredentialCreate @@ -195,6 +199,13 @@ async def lifespan(app: FastAPI): except Exception as e: print(f"Error initializing scheduler: {e}") + # Initialize log archive service + try: + log_archive_service.start() + print("Log archive service started") + except Exception as e: + print(f"Error starting log archive service: {e}") + yield # Shutdown the scheduler @@ -204,6 +215,13 @@ async def lifespan(app: FastAPI): except Exception as e: print(f"Error shutting down scheduler: {e}") + # Shutdown log archive service + try: + log_archive_service.stop() + print("Log archive service stopped") + except Exception as e: + print(f"Error stopping log archive service: {e}") + # Create FastAPI app app = FastAPI( @@ -242,6 +260,7 @@ async def lifespan(app: FastAPI): app.include_router(public_router) app.include_router(setup_router) app.include_router(telemetry_router) +app.include_router(logs_router) # WebSocket endpoint diff --git a/src/backend/routers/logs.py b/src/backend/routers/logs.py new file mode 100644 index 000000000..588da50e7 --- /dev/null +++ b/src/backend/routers/logs.py @@ -0,0 +1,189 @@ +""" +Log management API endpoints. + +Provides admin-only access to log statistics, retention configuration, +and manual archival operations. +""" +from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel, Field +from typing import Optional + +from models import User +from dependencies import get_current_user +from services.log_archive_service import log_archive_service, LOG_RETENTION_DAYS +import logging + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/logs", tags=["logs"]) + + +# Request/Response Models + +class RetentionConfig(BaseModel): + """Retention configuration.""" + retention_days: int = Field(..., ge=1, le=3650, description="Days to retain logs") + archive_enabled: bool = Field(..., description="Whether archival is enabled") + cleanup_hour: int = Field(..., ge=0, le=23, description="Hour (UTC) to run nightly archival") + + +class ArchiveRequest(BaseModel): + """Manual archive request.""" + retention_days: Optional[int] = Field(None, ge=1, le=3650, description="Override retention days") + delete_after_archive: bool = Field(True, description="Delete originals after archiving") + + +# Dependencies + +def require_admin(current_user: User = Depends(get_current_user)): + """Require admin role for log management.""" + if current_user.role != "admin": + raise HTTPException(status_code=403, detail="Admin access required") + return current_user + + +# Endpoints + +@router.get("/stats") +async def get_log_stats(current_user: User = Depends(require_admin)): + """ + Get statistics about log files. + + Returns information about active log files and archives including + sizes, counts, and date ranges. + """ + try: + stats = log_archive_service.get_log_stats() + return { + "log_files": stats["log_files"], + "archive_files": stats["archive_files"], + "total_log_size": stats["total_log_size"], + "total_log_size_mb": round(stats["total_log_size"] / 1024 / 1024, 2), + "total_archive_size": stats["total_archive_size"], + "total_archive_size_mb": round(stats["total_archive_size"] / 1024 / 1024, 2), + "oldest_log": stats["oldest_log"], + "newest_log": stats["newest_log"], + "log_file_count": len(stats["log_files"]), + "archive_file_count": len(stats["archive_files"]), + } + except Exception as e: + logger.error(f"Failed to get log stats: {e}") + raise HTTPException(status_code=500, detail=f"Failed to get log stats: {str(e)}") + + +@router.get("/retention", response_model=RetentionConfig) +async def get_retention_config(current_user: User = Depends(require_admin)): + """ + Get current retention configuration. + + Returns the active retention policy settings. + """ + import os + + return RetentionConfig( + retention_days=int(os.getenv("LOG_RETENTION_DAYS", "90")), + archive_enabled=os.getenv("LOG_ARCHIVE_ENABLED", "true").lower() == "true", + cleanup_hour=int(os.getenv("LOG_CLEANUP_HOUR", "3")), + ) + + +@router.put("/retention", response_model=RetentionConfig) +async def update_retention_config( + config: RetentionConfig, + current_user: User = Depends(require_admin) +): + """ + Update retention configuration. + + Note: This only updates the runtime configuration. To persist changes, + update the environment variables in docker-compose.yml or .env file. + """ + import os + + # Update runtime environment variables + os.environ["LOG_RETENTION_DAYS"] = str(config.retention_days) + os.environ["LOG_ARCHIVE_ENABLED"] = str(config.archive_enabled).lower() + os.environ["LOG_CLEANUP_HOUR"] = str(config.cleanup_hour) + + logger.info( + f"Retention config updated by {current_user.username}: " + f"{config.retention_days} days, archive={config.archive_enabled}" + ) + + # Reschedule the job with new cleanup hour if scheduler is running + if log_archive_service.scheduler.running: + try: + # Remove and re-add the job with new hour + log_archive_service.scheduler.remove_job("log_archival") + except Exception: + pass # Job might not exist + + from apscheduler.triggers.cron import CronTrigger + log_archive_service.scheduler.add_job( + log_archive_service.archive_old_logs, + CronTrigger(hour=config.cleanup_hour, minute=0), + id="log_archival", + name="Nightly Log Archival", + replace_existing=True, + misfire_grace_time=3600, + ) + + return config + + +@router.post("/archive") +async def trigger_archive( + request: ArchiveRequest = ArchiveRequest(), + current_user: User = Depends(require_admin) +): + """ + Manually trigger log archival. + + Archives log files older than the specified retention period. + Optionally override the retention days for this operation. + """ + try: + logger.info( + f"Manual archive triggered by {current_user.username} " + f"(retention_days={request.retention_days or LOG_RETENTION_DAYS})" + ) + + result = await log_archive_service.archive_old_logs( + retention_days=request.retention_days, + delete_after_archive=request.delete_after_archive, + ) + + return { + "status": "success", + "files_processed": result["files_processed"], + "bytes_before": result["bytes_before"], + "bytes_after": result["bytes_after"], + "bytes_saved": result["bytes_saved"], + "bytes_saved_mb": round(result["bytes_saved"] / 1024 / 1024, 2), + "files_stored": result["files_stored"], + "errors": result["errors"], + "retention_days": result["retention_days"], + "cutoff_date": result["cutoff_date"], + } + + except Exception as e: + logger.error(f"Archive failed: {e}") + raise HTTPException(status_code=500, detail=f"Archive failed: {str(e)}") + + +@router.get("/health") +async def log_service_health(current_user: User = Depends(require_admin)): + """ + Get health status of log archival service. + + Returns information about the scheduler and storage configuration. + """ + import os + + return { + "scheduler_running": log_archive_service.scheduler.running if log_archive_service.scheduler else False, + "archive_enabled": os.getenv("LOG_ARCHIVE_ENABLED", "true").lower() == "true", + "archive_path": os.getenv("LOG_ARCHIVE_PATH", "/data/archives"), + "retention_days": int(os.getenv("LOG_RETENTION_DAYS", "90")), + "cleanup_hour": int(os.getenv("LOG_CLEANUP_HOUR", "3")), + } + diff --git a/src/backend/services/archive_storage.py b/src/backend/services/archive_storage.py new file mode 100644 index 000000000..a67289f7f --- /dev/null +++ b/src/backend/services/archive_storage.py @@ -0,0 +1,221 @@ +""" +Abstract storage interface for log archives. + +Supports pluggable backends with local-only default for sovereign deployments. +""" +import os +import shutil +import logging +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Optional, Dict, List, Any +from datetime import datetime + +logger = logging.getLogger(__name__) + + +class ArchiveStorage(ABC): + """Abstract base class for archive storage backends.""" + + @abstractmethod + async def store_archive( + self, + source_path: Path, + metadata: Optional[Dict[str, str]] = None + ) -> Dict[str, Any]: + """ + Store an archive file. + + Args: + source_path: Path to the archive file to store + metadata: Optional metadata to attach to the archive + + Returns: + Dict with storage info (path, size, etc.) + + Raises: + Exception if storage fails + """ + pass + + @abstractmethod + def list_archives(self, prefix: str = "") -> List[Dict[str, Any]]: + """ + List stored archives. + + Args: + prefix: Optional prefix to filter archives + + Returns: + List of archive info dicts + """ + pass + + @abstractmethod + def delete_archive(self, archive_name: str): + """ + Delete a stored archive. + + Args: + archive_name: Name of the archive to delete + """ + pass + + +class LocalArchiveStorage(ArchiveStorage): + """ + Local filesystem storage for archives. + + Archives are moved to a specified local directory which can be: + - A Docker volume for persistence + - A mounted NAS/NFS share + - A network-attached storage device + """ + + def __init__(self, archive_path: str = "/data/archives"): + """ + Initialize local archive storage. + + Args: + archive_path: Directory path for storing archives + """ + self.archive_path = Path(archive_path) + self.archive_path.mkdir(parents=True, exist_ok=True) + logger.info(f"Local archive storage initialized: {self.archive_path}") + + async def store_archive( + self, + source_path: Path, + metadata: Optional[Dict[str, str]] = None + ) -> Dict[str, Any]: + """ + Move archive to local storage directory. + + Args: + source_path: Path to the archive file + metadata: Optional metadata (stored as JSON sidecar file) + + Returns: + Dict with storage location and size + + Raises: + Exception if move fails + """ + if not source_path.exists(): + raise FileNotFoundError(f"Archive file not found: {source_path}") + + try: + dest_path = self.archive_path / source_path.name + file_size = source_path.stat().st_size + + # Move file to archive directory (atomic on same filesystem) + shutil.move(str(source_path), str(dest_path)) + + logger.info( + f"Archived {source_path.name} to local storage " + f"({file_size / 1024 / 1024:.1f} MB)" + ) + + # Store metadata as sidecar JSON if provided + if metadata: + metadata_path = dest_path.with_suffix(dest_path.suffix + '.meta') + import json + metadata_path.write_text(json.dumps(metadata, indent=2)) + + return { + "storage_type": "local", + "path": str(dest_path), + "size": file_size, + "timestamp": datetime.utcnow().isoformat(), + } + + except Exception as e: + error_msg = f"Failed to store archive {source_path.name}: {e}" + logger.error(error_msg) + raise Exception(error_msg) + + def list_archives(self, prefix: str = "") -> List[Dict[str, Any]]: + """ + List archives in local storage. + + Args: + prefix: Filter archives by filename prefix + + Returns: + List of archive info dicts + """ + try: + archives = [] + pattern = f"{prefix}*.json.gz" if prefix else "*.json.gz" + + for archive_path in self.archive_path.glob(pattern): + stat = archive_path.stat() + + # Load metadata if available + metadata_path = archive_path.with_suffix(archive_path.suffix + '.meta') + metadata = {} + if metadata_path.exists(): + import json + try: + metadata = json.loads(metadata_path.read_text()) + except Exception as e: + logger.debug(f"Could not load metadata for {archive_path.name}: {e}") + + archives.append({ + "name": archive_path.name, + "path": str(archive_path), + "size": stat.st_size, + "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(), + "metadata": metadata, + }) + + return sorted(archives, key=lambda x: x["modified"], reverse=True) + + except Exception as e: + logger.error(f"Failed to list archives: {e}") + raise + + def delete_archive(self, archive_name: str): + """ + Delete an archive from local storage. + + Args: + archive_name: Name of the archive file to delete + """ + try: + archive_path = self.archive_path / archive_name + + if not archive_path.exists(): + logger.warning(f"Archive not found for deletion: {archive_name}") + return + + # Delete metadata sidecar if exists + metadata_path = archive_path.with_suffix(archive_path.suffix + '.meta') + if metadata_path.exists(): + metadata_path.unlink() + + # Delete archive file + archive_path.unlink() + logger.info(f"Deleted archive: {archive_name}") + + except Exception as e: + logger.error(f"Failed to delete archive {archive_name}: {e}") + raise + + +def get_archive_storage() -> ArchiveStorage: + """ + Factory function to get the configured archive storage backend. + + Currently only returns LocalArchiveStorage for sovereign deployment. + Future backends can be added here based on configuration. + + Returns: + Configured ArchiveStorage instance + """ + # Get archive path from environment + archive_path = os.getenv("LOG_ARCHIVE_PATH", "/data/archives") + + logger.info("Using local archive storage (sovereign mode)") + return LocalArchiveStorage(archive_path=archive_path) + diff --git a/src/backend/services/log_archive_service.py b/src/backend/services/log_archive_service.py new file mode 100644 index 000000000..55e8dfa88 --- /dev/null +++ b/src/backend/services/log_archive_service.py @@ -0,0 +1,291 @@ +""" +Log Archive Service for Trinity Vector Logs. + +Handles automatic archival of old log files with compression. +""" +import os +import gzip +import shutil +import hashlib +from datetime import datetime, timedelta +from pathlib import Path +from typing import Optional, Dict, Any, List +import asyncio +import logging + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.cron import CronTrigger + +from .archive_storage import get_archive_storage + +logger = logging.getLogger(__name__) + +# Configuration from environment +LOG_RETENTION_DAYS = int(os.getenv("LOG_RETENTION_DAYS", "90")) +LOG_ARCHIVE_ENABLED = os.getenv("LOG_ARCHIVE_ENABLED", "true").lower() == "true" +LOG_CLEANUP_HOUR = int(os.getenv("LOG_CLEANUP_HOUR", "3")) + +LOG_DIR = Path("/data/logs") +TEMP_ARCHIVE_DIR = Path("/tmp/archives") + + +class LogArchiveService: + """Service for archiving old Vector log files.""" + + def __init__(self): + self.scheduler = AsyncIOScheduler() + self.storage = get_archive_storage() + logger.info("Archive storage initialized") + + def start(self): + """Start the archival scheduler.""" + if not LOG_ARCHIVE_ENABLED: + logger.info("Log archival disabled (LOG_ARCHIVE_ENABLED=false)") + return + + # Ensure temp archive directory exists for compression staging + TEMP_ARCHIVE_DIR.mkdir(parents=True, exist_ok=True) + + # Schedule nightly archival + self.scheduler.add_job( + self.archive_old_logs, + CronTrigger(hour=LOG_CLEANUP_HOUR, minute=0), + id="log_archival", + name="Nightly Log Archival", + replace_existing=True, + misfire_grace_time=3600, # 1 hour grace period + ) + + self.scheduler.start() + logger.info( + f"Log archival scheduler started: Daily at {LOG_CLEANUP_HOUR:02d}:00 UTC " + f"(retention: {LOG_RETENTION_DAYS} days)" + ) + + def stop(self): + """Stop the archival scheduler.""" + if self.scheduler.running: + self.scheduler.shutdown(wait=False) + logger.info("Log archival scheduler stopped") + + async def archive_old_logs( + self, retention_days: Optional[int] = None, delete_after_archive: bool = True + ) -> Dict[str, Any]: + """ + Archive log files older than retention period. + + Args: + retention_days: Override default retention days + delete_after_archive: Delete originals after successful archive + + Returns: + Dict with archive statistics + """ + if retention_days is None: + retention_days = LOG_RETENTION_DAYS + + cutoff_date = datetime.utcnow() - timedelta(days=retention_days) + + if not LOG_DIR.exists(): + logger.warning(f"Log directory not found: {LOG_DIR}") + return {"error": "Log directory not found"} + + files_processed = 0 + bytes_before = 0 + bytes_after = 0 + files_stored = 0 + errors = [] + + # Find all log files older than retention period + for log_file in LOG_DIR.glob("*.json"): + try: + # Extract date from filename: platform-2025-10-06.json or agents-2025-10-06.json + file_date = self._extract_date_from_filename(log_file.name) + + if file_date and file_date < cutoff_date.date(): + # Compress the file to temp directory + archive_path = TEMP_ARCHIVE_DIR / f"{log_file.stem}.json.gz" + original_size = log_file.stat().st_size + + logger.info(f"Archiving {log_file.name} ({original_size / 1024 / 1024:.1f} MB)") + + # Compress with integrity check + if await self._compress_file(log_file, archive_path): + compressed_size = archive_path.stat().st_size + bytes_before += original_size + bytes_after += compressed_size + + # Store archive using storage backend + try: + metadata = { + "original_size": str(original_size), + "compressed_size": str(compressed_size), + "archived_at": datetime.utcnow().isoformat(), + "retention_days": str(retention_days), + "original_file": log_file.name, + } + + await self.storage.store_archive(archive_path, metadata) + files_stored += 1 + + except Exception as e: + logger.error(f"Storage failed for {archive_path.name}: {e}") + errors.append(f"Storage failed: {archive_path.name}") + # Clean up temp file + if archive_path.exists(): + archive_path.unlink() + continue + + # Delete original if requested + if delete_after_archive: + log_file.unlink() + logger.info(f"Deleted original: {log_file.name}") + + files_processed += 1 + else: + errors.append(f"Compression failed: {log_file.name}") + + except Exception as e: + logger.error(f"Error archiving {log_file.name}: {e}") + errors.append(f"Error: {log_file.name} - {str(e)}") + + # Log summary + bytes_saved = bytes_before - bytes_after + logger.info( + f"Archival complete: {files_processed} files, " + f"{bytes_saved / 1024 / 1024:.1f} MB saved " + f"({files_stored} files stored)" + ) + + return { + "files_processed": files_processed, + "bytes_before": bytes_before, + "bytes_after": bytes_after, + "bytes_saved": bytes_saved, + "files_stored": files_stored, + "errors": errors, + "retention_days": retention_days, + "cutoff_date": cutoff_date.isoformat(), + } + + async def _compress_file(self, source: Path, dest: Path) -> bool: + """ + Compress a file with gzip and verify integrity. + + Args: + source: Source file path + dest: Destination archive path + + Returns: + True if compression and verification succeeded + """ + try: + # Calculate checksum of original + original_checksum = self._calculate_checksum(source) + + # Compress using gzip level 9 + with open(source, 'rb') as f_in: + with gzip.open(dest, 'wb', compresslevel=9) as f_out: + shutil.copyfileobj(f_in, f_out, length=1024*1024) # 1MB chunks + + # Verify by decompressing and checking checksum + with gzip.open(dest, 'rb') as f: + data = f.read() + decompressed_checksum = hashlib.sha256(data).hexdigest() + + if original_checksum != decompressed_checksum: + logger.error(f"Integrity check failed for {dest}") + dest.unlink() # Delete corrupt archive + return False + + return True + + except Exception as e: + logger.error(f"Compression failed for {source}: {e}") + if dest.exists(): + dest.unlink() + return False + + def _calculate_checksum(self, file_path: Path) -> str: + """Calculate SHA256 checksum of a file.""" + sha256 = hashlib.sha256() + with open(file_path, 'rb') as f: + for chunk in iter(lambda: f.read(1024*1024), b''): + sha256.update(chunk) + return sha256.hexdigest() + + def _extract_date_from_filename(self, filename: str) -> Optional[datetime.date]: + """ + Extract date from log filename. + + Expected format: platform-2025-10-06.json or agents-2025-10-06.json + + Args: + filename: Log filename + + Returns: + datetime.date object or None if parsing fails + """ + try: + # Remove extension and split by hyphen + parts = filename.replace('.json', '').split('-') + + # Last 3 parts should be YYYY-MM-DD + if len(parts) >= 4: + year = int(parts[-3]) + month = int(parts[-2]) + day = int(parts[-1]) + return datetime(year, month, day).date() + except (ValueError, IndexError): + logger.debug(f"Could not extract date from filename: {filename}") + + return None + + def get_log_stats(self) -> Dict[str, Any]: + """Get statistics about log files.""" + stats = { + "log_files": [], + "archive_files": [], + "total_log_size": 0, + "total_archive_size": 0, + "oldest_log": None, + "newest_log": None, + } + + if LOG_DIR.exists(): + log_files = list(LOG_DIR.glob("*.json")) + for log_file in log_files: + file_size = log_file.stat().st_size + file_date = self._extract_date_from_filename(log_file.name) + + stats["log_files"].append({ + "name": log_file.name, + "size": file_size, + "date": file_date.isoformat() if file_date else None, + }) + stats["total_log_size"] += file_size + + if file_date: + if not stats["oldest_log"] or file_date < datetime.fromisoformat(stats["oldest_log"]).date(): + stats["oldest_log"] = file_date.isoformat() + if not stats["newest_log"] or file_date > datetime.fromisoformat(stats["newest_log"]).date(): + stats["newest_log"] = file_date.isoformat() + + # Get archived files from storage backend + try: + archives = self.storage.list_archives() + for archive in archives: + stats["archive_files"].append({ + "name": archive["name"], + "size": archive["size"], + }) + stats["total_archive_size"] += archive["size"] + except Exception as e: + logger.error(f"Failed to list archives: {e}") + + return stats + + +# Global instance +log_archive_service = LogArchiveService() + diff --git a/tests/requirements-test.txt b/tests/requirements-test.txt index 393cb1f17..17691e7d5 100644 --- a/tests/requirements-test.txt +++ b/tests/requirements-test.txt @@ -6,3 +6,8 @@ pytest-html>=4.1.0 pytest-cov>=6.0.0 httpx>=0.28.0 python-dotenv>=1.0.1 +docker>=7.1.0 +apscheduler>=3.11.0 +pydantic>=2.10.0 +fastapi>=0.115.0 +pyyaml>=6.0.0 diff --git a/tests/test_log_archive.py b/tests/test_log_archive.py new file mode 100644 index 000000000..9c7410c30 --- /dev/null +++ b/tests/test_log_archive.py @@ -0,0 +1,509 @@ +""" +Tests for Vector log archival and retention. + +Tests the log archive service, API endpoints, compression, and local storage. +""" +import pytest +import json +import gzip +import sys +from datetime import datetime, timedelta +from pathlib import Path +from unittest.mock import Mock, patch, AsyncMock +import tempfile +import shutil + +# Add backend to path for service imports +sys.path.insert(0, str(Path(__file__).parent.parent / "src" / "backend")) + +from utils.assertions import assert_status, assert_status_in + + +# Test fixtures + +@pytest.fixture +def test_log_dir(tmp_path): + """Create a temporary log directory for testing.""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + return log_dir + + +@pytest.fixture +def test_archive_dir(tmp_path): + """Create a temporary archive directory for testing.""" + archive_dir = tmp_path / "archives" + archive_dir.mkdir() + return archive_dir + + +@pytest.fixture +def mock_log_files(test_log_dir): + """Create mock log files with different dates.""" + files = [] + + # Create files for different dates + dates = [ + datetime.utcnow().date(), # Today + (datetime.utcnow() - timedelta(days=30)).date(), # 30 days ago + (datetime.utcnow() - timedelta(days=95)).date(), # 95 days ago + (datetime.utcnow() - timedelta(days=180)).date(), # 180 days ago + ] + + for i, date in enumerate(dates): + # Create platform and agent log files + for log_type in ["platform", "agents"]: + file_name = f"{log_type}-{date.isoformat()}.json" + file_path = test_log_dir / file_name + + # Write sample log entries + log_entries = [ + { + "timestamp": f"{date}T{hour:02d}:00:00Z", + "level": "info", + "message": f"Test log entry {i}-{hour}", + "container_name": f"trinity-backend" if log_type == "platform" else f"agent-test", + } + for hour in range(24) + ] + + with open(file_path, 'w') as f: + for entry in log_entries: + f.write(json.dumps(entry) + '\n') + + files.append(file_path) + + return files + + +# ========================================================================= +# Test: API Authentication +# ========================================================================= + +class TestLogArchiveAuthentication: + """Test that all log endpoints require admin authentication.""" + + def test_stats_requires_auth(self, unauthenticated_client): + """Test GET /api/logs/stats requires authentication.""" + response = unauthenticated_client.get("/api/logs/stats", auth=False) + assert_status(response, 401) + + def test_retention_config_requires_auth(self, unauthenticated_client): + """Test GET /api/logs/retention requires authentication.""" + response = unauthenticated_client.get("/api/logs/retention", auth=False) + assert_status(response, 401) + + def test_update_retention_requires_auth(self, unauthenticated_client): + """Test PUT /api/logs/retention requires authentication.""" + response = unauthenticated_client.put( + "/api/logs/retention", + json={"retention_days": 30, "archive_enabled": True, "cleanup_hour": 3}, + auth=False + ) + assert_status(response, 401) + + def test_archive_requires_auth(self, unauthenticated_client): + """Test POST /api/logs/archive requires authentication.""" + response = unauthenticated_client.post("/api/logs/archive", json={}, auth=False) + assert_status(response, 401) + + def test_health_requires_auth(self, unauthenticated_client): + """Test GET /api/logs/health requires authentication.""" + response = unauthenticated_client.get("/api/logs/health", auth=False) + assert_status(response, 401) + + +# ========================================================================= +# Test: Stats Endpoint +# ========================================================================= + +class TestLogStats: + """Test log statistics endpoint.""" + + def test_stats_returns_expected_fields(self, api_client): + """Test that stats endpoint returns expected structure.""" + response = api_client.get("/api/logs/stats") + + assert_status(response, 200) + data = response.json() + + # Check required fields + assert "log_files" in data + assert "archive_files" in data + assert "total_log_size" in data + assert "total_log_size_mb" in data + assert "total_archive_size" in data + assert "total_archive_size_mb" in data + assert "log_file_count" in data + assert "archive_file_count" in data + + def test_stats_log_files_structure(self, api_client): + """Test that log files array has expected structure.""" + response = api_client.get("/api/logs/stats") + + assert_status(response, 200) + data = response.json() + + if data["log_file_count"] > 0: + log_file = data["log_files"][0] + assert "name" in log_file + assert "size" in log_file + assert "date" in log_file or log_file["date"] is None + + +# ========================================================================= +# Test: Retention Configuration +# ========================================================================= + +class TestRetentionConfig: + """Test retention configuration endpoints.""" + + def test_get_retention_returns_defaults(self, api_client): + """Test GET /api/logs/retention returns default values.""" + response = api_client.get("/api/logs/retention") + + assert_status(response, 200) + data = response.json() + + assert "retention_days" in data + assert "archive_enabled" in data + assert "cleanup_hour" in data + + assert isinstance(data["retention_days"], int) + assert isinstance(data["archive_enabled"], bool) + assert isinstance(data["cleanup_hour"], int) + assert 0 <= data["cleanup_hour"] <= 23 + + def test_update_retention_validates_days_minimum(self, api_client): + """Test that retention days must be at least 1.""" + response = api_client.put( + "/api/logs/retention", + json={"retention_days": 0, "archive_enabled": True, "cleanup_hour": 3} + ) + + assert_status_in(response, [400, 422]) + + def test_update_retention_validates_days_maximum(self, api_client): + """Test that retention days cannot exceed 3650.""" + response = api_client.put( + "/api/logs/retention", + json={"retention_days": 9999, "archive_enabled": True, "cleanup_hour": 3} + ) + + assert_status_in(response, [400, 422]) + + def test_update_retention_validates_hour_range(self, api_client): + """Test that cleanup hour must be 0-23.""" + response = api_client.put( + "/api/logs/retention", + json={"retention_days": 90, "archive_enabled": True, "cleanup_hour": 25} + ) + + assert_status_in(response, [400, 422]) + + def test_update_retention_succeeds(self, api_client): + """Test successful retention update.""" + response = api_client.put( + "/api/logs/retention", + json={"retention_days": 30, "archive_enabled": True, "cleanup_hour": 2} + ) + + assert_status(response, 200) + data = response.json() + + assert data["retention_days"] == 30 + assert data["archive_enabled"] is True + assert data["cleanup_hour"] == 2 + + +# ========================================================================= +# Test: Archive Service +# ========================================================================= + +class TestArchiveService: + """Test log archive service functionality.""" + + @pytest.mark.asyncio + async def test_extract_date_from_filename(self): + """Test date extraction from log filenames.""" + from services.log_archive_service import LogArchiveService + + service = LogArchiveService() + + # Valid filenames + assert service._extract_date_from_filename("platform-2025-10-06.json") == datetime(2025, 10, 6).date() + assert service._extract_date_from_filename("agents-2026-01-05.json") == datetime(2026, 1, 5).date() + + # Invalid filenames + assert service._extract_date_from_filename("platform.json") is None + assert service._extract_date_from_filename("invalid-name.json") is None + assert service._extract_date_from_filename("platform-2025-99-99.json") is None + + @pytest.mark.asyncio + async def test_compression_reduces_size(self, test_log_dir, test_archive_dir): + """Test that compression significantly reduces file size.""" + from services.log_archive_service import LogArchiveService + + # Create a large test file + test_file = test_log_dir / "platform-2025-01-01.json" + with open(test_file, 'w') as f: + for i in range(10000): + f.write(json.dumps({ + "timestamp": f"2025-01-01T{i % 24:02d}:00:00Z", + "message": f"Test log entry {i}" * 10, + "level": "info" + }) + '\n') + + original_size = test_file.stat().st_size + archive_path = test_archive_dir / "platform-2025-01-01.json.gz" + + service = LogArchiveService() + success = await service._compress_file(test_file, archive_path) + + assert success + assert archive_path.exists() + + compressed_size = archive_path.stat().st_size + compression_ratio = compressed_size / original_size + + # Should achieve at least 50% compression + assert compression_ratio < 0.5, f"Compression ratio {compression_ratio:.2%} is too high" + + @pytest.mark.asyncio + async def test_compression_integrity_check(self, test_log_dir, test_archive_dir): + """Test that compression integrity check works.""" + from services.log_archive_service import LogArchiveService + + test_file = test_log_dir / "platform-2025-01-01.json" + with open(test_file, 'w') as f: + f.write(json.dumps({"test": "data"}) + '\n') + + archive_path = test_archive_dir / "platform-2025-01-01.json.gz" + + service = LogArchiveService() + success = await service._compress_file(test_file, archive_path) + + assert success + + # Verify we can decompress and get original content + with gzip.open(archive_path, 'rb') as f: + decompressed = f.read() + + with open(test_file, 'rb') as f: + original = f.read() + + assert decompressed == original + + +# ========================================================================= +# Test: Manual Archive Endpoint +# ========================================================================= + +class TestManualArchive: + """Test manual archive trigger endpoint.""" + + def test_archive_endpoint_exists(self, api_client): + """Test that archive endpoint is available.""" + response = api_client.post( + "/api/logs/archive", + json={"retention_days": 365, "delete_after_archive": False} + ) + + # Should succeed or return an expected error (not 404) + assert response.status_code != 404 + + def test_archive_returns_expected_fields(self, api_client): + """Test that archive response has expected structure.""" + response = api_client.post( + "/api/logs/archive", + json={"retention_days": 365, "delete_after_archive": False} + ) + + assert_status(response, 200) + data = response.json() + + assert "status" in data + assert "files_processed" in data + assert "bytes_saved" in data + assert "files_stored" in data + assert "retention_days" in data + + def test_archive_respects_retention_days_param(self, api_client): + """Test that archive respects custom retention_days parameter.""" + response = api_client.post( + "/api/logs/archive", + json={"retention_days": 30, "delete_after_archive": False} + ) + + assert_status(response, 200) + data = response.json() + + assert data["retention_days"] == 30 + + +# ========================================================================= +# Test: Health Endpoint +# ========================================================================= + +class TestLogServiceHealth: + """Test log service health endpoint.""" + + def test_health_returns_scheduler_status(self, api_client): + """Test that health endpoint returns scheduler status.""" + response = api_client.get("/api/logs/health") + + assert_status(response, 200) + data = response.json() + + assert "scheduler_running" in data + assert "archive_enabled" in data + assert "retention_days" in data + assert "cleanup_hour" in data + + assert isinstance(data["scheduler_running"], bool) + assert isinstance(data["archive_enabled"], bool) + + +# ========================================================================= +# Test: Integration +# ========================================================================= + +class TestLogArchiveIntegration: + """Integration tests for log archival workflow.""" + + @pytest.mark.asyncio + async def test_full_archive_workflow(self, test_log_dir, test_archive_dir): + """Test complete archive workflow with file operations.""" + from services.log_archive_service import LogArchiveService + from services.archive_storage import LocalArchiveStorage + + # Patch the service to use test directories + with patch('services.log_archive_service.LOG_DIR', test_log_dir), \ + patch('services.log_archive_service.TEMP_ARCHIVE_DIR', test_log_dir / "temp"): + + # Create temp directory + (test_log_dir / "temp").mkdir(exist_ok=True) + + # Create old log file + old_date = (datetime.utcnow() - timedelta(days=100)).date() + old_file = test_log_dir / f"platform-{old_date.isoformat()}.json" + + with open(old_file, 'w') as f: + for i in range(100): + f.write(json.dumps({"message": f"Log {i}"}) + '\n') + + # Create recent log file + recent_date = datetime.utcnow().date() + recent_file = test_log_dir / f"platform-{recent_date.isoformat()}.json" + + with open(recent_file, 'w') as f: + f.write(json.dumps({"message": "Recent log"}) + '\n') + + # Run archival with 90-day retention + service = LogArchiveService() + service.storage = LocalArchiveStorage(archive_path=str(test_archive_dir)) + + result = await service.archive_old_logs( + retention_days=90, + delete_after_archive=True + ) + + # Verify results + assert result["files_processed"] >= 1 + assert result["bytes_saved"] > 0 + + # Old file should be archived and deleted + assert not old_file.exists() + assert (test_archive_dir / f"platform-{old_date.isoformat()}.json.gz").exists() + + # Recent file should remain + assert recent_file.exists() + + def test_retention_config_persists_runtime_changes(self, api_client): + """Test that runtime retention changes persist until restart.""" + # Update retention + update_response = api_client.put( + "/api/logs/retention", + json={"retention_days": 45, "archive_enabled": True, "cleanup_hour": 4} + ) + + assert_status(update_response, 200) + + # Verify change persisted + get_response = api_client.get("/api/logs/retention") + + assert_status(get_response, 200) + data = get_response.json() + + assert data["retention_days"] == 45 + assert data["cleanup_hour"] == 4 + + +# ========================================================================= +# Test: Validation +# ========================================================================= + +class TestArchiveValidation: + """Test validation of archive parameters.""" + + def test_archive_validates_retention_days_negative(self, api_client): + """Test that negative retention days are rejected.""" + response = api_client.post( + "/api/logs/archive", + json={"retention_days": -1, "delete_after_archive": True} + ) + + assert_status_in(response, [400, 422]) + + def test_archive_validates_retention_days_excessive(self, api_client): + """Test that excessive retention days are rejected.""" + response = api_client.post( + "/api/logs/archive", + json={"retention_days": 5000, "delete_after_archive": True} + ) + + assert_status_in(response, [400, 422]) + + +# ========================================================================= +# Test: Error Handling +# ========================================================================= + +class TestArchiveErrorHandling: + """Test error handling in archive service.""" + + @pytest.mark.asyncio + async def test_archive_handles_missing_log_directory(self): + """Test graceful handling when log directory doesn't exist.""" + from services.log_archive_service import LogArchiveService + + with patch('services.log_archive_service.LOG_DIR', Path("/nonexistent/path")): + service = LogArchiveService() + result = await service.archive_old_logs() + + assert "error" in result + assert "not found" in result["error"].lower() + + @pytest.mark.asyncio + async def test_checksum_calculation(self): + """Test file checksum calculation.""" + from services.log_archive_service import LogArchiveService + + service = LogArchiveService() + + # Create test file + with tempfile.NamedTemporaryFile(mode='w', delete=False) as f: + f.write("test content") + test_file = Path(f.name) + + try: + checksum = service._calculate_checksum(test_file) + + # Should return valid SHA256 hex string + assert isinstance(checksum, str) + assert len(checksum) == 64 + assert all(c in '0123456789abcdef' for c in checksum) + + finally: + test_file.unlink() +