--- title: 'DE Upload Endpoint — Airflow Plugin' slug: 'de-upload-endpoint-airflow-plugin' created: '2026-03-16' status: 'done' stepsCompleted: [1, 2, 3, 4] tech_stack: ['Apache Airflow 3.0.6', 'Flask Blueprint (verified supported in 3.0.6)', 'Python 3.11', 'boto3 (s3v4 signature)', 'SQLAlchemy 2.x (sync for plugin, async for tasks)', 'PostgreSQL 16', 'SeaweedFS (S3-compatible, port 8333)'] files_to_modify: - 'airflow/plugins/__init__.py' - 'airflow/plugins/api/__init__.py (NEW)' - 'airflow/plugins/api/upload_blueprint.py (NEW)' - 'airflow/plugins/api/auth.py (NEW)' - 'src/core/config.py (add upload_bucket, max_filename_length, HMAC settings)' - 'airflow/plugins/operators/validate_event.py (add .ppt/.pptm extensions)' - 'docker/docker-compose.airflow.yml (env var passthrough)' - '.env.example (add new env vars)' code_patterns: - 'AirflowPlugin with flask_blueprints=[blueprint] registration' - 'Flask Blueprint for custom REST endpoint on Airflow webserver (port 8081→8080)' - 'boto3 S3 client: Config(signature_version=s3v4), region_name=us-east-1, reads AWS_* env vars' - 'Sync SQLAlchemy engine for Flask plugin (create_engine, not create_async_engine)' - 'Full HMAC-SHA256 verification per SPEC-HMAC Section 3.3 (7 headers, key ring, constant-time compare)' - 'Airflow Python API: trigger_dag or DagRun.create_dagrun (internal, no auth, import resolution at startup)' - 'Filename sanitization: os.path.basename + strip null/traversal (preserves CJK/Unicode)' - 'Flask MAX_CONTENT_LENGTH for early rejection of oversized uploads' - 'validate_s3_path(tenant_id, node_id, s3_key) for cross-tenant S3 protection' - 'Document model: TenantIsolationMixin provides tenant_id/node_id, api_key_id nullable' - 'File validation: FileValidationError(code, message) with INVALID_EXTENSION/FILE_TOO_LARGE/INVALID_FORMAT' test_patterns: [] --- # Tech-Spec: DE Upload Endpoint — Airflow Plugin **Created:** 2026-03-16 ## Overview ### Problem Statement The current FastAPI app (`src/api/`) is deprecated. There is no upload endpoint in the Airflow deployment for BFF services (APA/UPA) to proxy file uploads to DE. The only way to trigger the `document_extraction_dag` is via the CLI script `scripts/upload_documents.py` or direct Airflow REST API calls — neither supports receiving multipart file uploads from other services. Per data-isolation-spec Section 6.4, DE must own the entire write path: receive raw files, write to SeaweedFS, create the DB record, and trigger the Airflow DAG. Currently APA writes files to SeaweedFS directly and passes the URL to DE, violating the CQRS single-writer principle. ### Solution Create an Airflow Flask Blueprint plugin that exposes `POST /api/v1/documents/upload` on the Airflow webserver (port 8081). The endpoint: 1. Validates HMAC headers (`X-Tenant-ID`, `X-Node-ID`) — extracts tenant/node context 2. Validates the uploaded file (extension, size, magic bytes) 3. Creates a `documents` record in PostgreSQL (status=pending, with tenant/node) 4. Writes the file to SeaweedFS `source-files` bucket with tenant-scoped key `{tenant_id}/{node_id}/{document_id}/{filename}` 5. Triggers `document_extraction_dag` via Airflow Python API with `conf: {document_id, object_key, tenant_id, node_id}` 6. Returns `{document_id, filename, status: "pending"}` to the caller ### Scope **In Scope:** - Airflow Flask Blueprint plugin with `POST /api/v1/documents/upload` endpoint - Full HMAC-SHA256 header verification per SPEC-HMAC (7 headers, key ring, signature verification, timestamp check) - SeaweedFS file upload with tenant-scoped key `{tenant_id}/{node_id}/{document_id}/{filename}` - PostgreSQL `documents` record creation (status=pending, with tenant_id/node_id) - Trigger `document_extraction_dag` via Airflow Python API (`DagBag` or `trigger_dag_run`) - File validation: extension whitelist, 50MB size limit, magic byte check - `validate_s3_path()` enforcement on uploaded key **Out of Scope:** - Deprecating/removing the FastAPI app (separate task) - Key ring rotation automation (manual rotation supported; auto-rotation via Vault watcher is future work) - Batch upload (single file per request for v1) - APA refactoring to call this endpoint (APA team task) - Status polling endpoint (use existing Airflow REST API `/api/v2/dags/.../dagRuns/{id}`) - File download / BFF proxy endpoint (separate spec) ## Context for Development ### Codebase Patterns **Airflow 3.0.6 Plugin Architecture (verified):** - Base image: `apache/airflow:3.0.6` (from `docker/Dockerfile.airflow:1`) - `airflow/plugins/__init__.py`: `TextIQPlugin(AirflowPlugin)` with `name = "textiq"`. Currently no `flask_blueprints` attribute. - Flask blueprints **confirmed supported** in Airflow 3.0.6 — register via `flask_blueprints = [blueprint]` on the `AirflowPlugin` class. - Airflow 3.x has hybrid architecture: Flask webserver (port 8080) + FastAPI API server. Blueprint plugins attach to the Flask webserver. - Plugin auto-discovered from `/opt/airflow/plugins/`. - Auth: `SimpleAuthManager` with all admins (dev mode). Blueprint endpoints bypass Airflow auth — must implement own auth via HMAC headers. **SeaweedFS Client (reuse existing pattern):** - `src/services/seaweedfs_service.py:44-61`: `boto3.client('s3', config=Config(signature_version='s3v4'), region_name='us-east-1')` - Container env vars (from `docker-compose.airflow.yml:11-14`): - `AWS_ACCESS_KEY_ID=${SEAWEEDFS_ACCESS_KEY_ID:-dummy}` - `AWS_SECRET_ACCESS_KEY=${SEAWEEDFS_SECRET_ACCESS_KEY:-dummy}` - `AWS_ENDPOINT_URL=http://seaweedfs:${SEAWEEDFS_PORT:-8333}` - boto3 reads `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` automatically. Must pass `endpoint_url` explicitly from `AWS_ENDPOINT_URL` env var. - `validate_s3_path(tenant_id, node_id, s3_key)` raises `PermissionError` if key doesn't match `{tenant_id}/{node_id}/` prefix. **Settings & DB (reuse `src/core/config.py`):** - `src/core/config.py` has `Settings(BaseSettings)` with `database_url`, `max_upload_size_mb`, `milvus_collection`, etc. Reads from `.env` automatically via `SettingsConfigDict(env_file=".env")`. - `settings.sync_database_url` property already converts `+asyncpg` → `+psycopg2` — use this directly for sync engine. No manual URL replacement needed. - `settings.max_upload_size_mb` already exists (default: 50) — use for file size validation and Flask `MAX_CONTENT_LENGTH`. - Import: `from src.core.config import settings` (src is on PYTHONPATH). - `Document` model at `src/db/models.py:68` — required fields: `filename`, `file_path`, `file_size_bytes`, `status`, `tenant_id` (UUID), `node_id` (UUID). `api_key_id` nullable. **File Validation (from `src/services/file_validator.py`):** - Extensions: `{".xlsx", ".xls", ".xlsm", ".docx", ".doc", ".docm", ".pdf", ".pptx", ".ppt", ".pptm"}` - Size: `MAX_FILE_SIZE_BYTES = 50 * 1024 * 1024` (50MB) - Magic bytes: `PK` (ZIP-based Office .xlsx/.docx/.pptx), `\xd0\xcf\x11\xe0` (OLE2 .xls/.doc), `%PDF-` (PDF) - `FileValidationError(code, message)` with codes: `INVALID_EXTENSION`, `FILE_TOO_LARGE`, `INVALID_FORMAT`, `CORRUPTED_FILE`, `EMPTY_FILE` - Can import directly: `from src.services.file_validator import ...` (src is on PYTHONPATH) **Password Detection (from `src/extraction_v2/password_detection.py`):** - `check_password_protected(file_path)` — raises `PasswordProtectedError` if file is encrypted - OOXML formats (.docx/.xlsx/.pptx/.docm/.xlsm/.pptm): normal files are ZIP (`PK` magic), encrypted files are OLE2 (`\xd0\xcf\x11\xe0`) — detect by checking magic bytes - Legacy formats (.doc/.xls/.ppt): always OLE2, check for `EncryptionInfo` stream, FIB fEncrypted bit (.doc), FilePass record (.xls) - Must run **after** saving file to temp location (needs file path, not stream) - Import: `from src.extraction_v2.password_detection import check_password_protected, PasswordProtectedError` **DAG Trigger (Python API — recommended for plugin):** - `from airflow.api.common.trigger_dag import trigger_dag` — runs inside webserver process, no HTTP call or JWT auth needed. - Alternative: `DagBag().get_dag('document_extraction_dag').create_dagrun(conf={...}, run_type='manual')` - DAG params: `document_id` (string UUID), `object_key` (S3 path), `tenant_id` (string UUID), `node_id` (string UUID) **Docker Environment:** - Airflow webserver: port 8081 (host) → 8080 (container), command: `webserver` - PYTHONPATH: `/opt/airflow/plugins:/opt/airflow/src:/opt/airflow` - `src` mounted read-only at `/opt/airflow/src:ro` - DB hostname: `postgres` (Docker network), not `localhost` - SeaweedFS hostname: `seaweedfs` (Docker network) - `psycopg2-binary` installed (for sync PostgreSQL access) ### Files to Reference | File | Purpose | | ---- | ------- | | `airflow/plugins/__init__.py` | Plugin registration — add `flask_blueprints` here | | `airflow/plugins/operators/validate_event.py` | EXTENSION_TO_CONTENT_TYPE map, validation patterns | | `src/services/seaweedfs_service.py` | boto3 S3 client pattern, `validate_s3_path()` | | `src/services/file_validator.py` | File validation logic (extensions, size, magic bytes) | | `src/extraction_v2/password_detection.py` | `check_password_protected()`, `PasswordProtectedError` — detects encrypted Office files | | `src/core/config.py` | `Settings` class — `sync_database_url`, `max_upload_size_mb`. Add `upload_bucket`, `max_filename_length`, HMAC settings here | | `src/db/models.py:68-109` | Document model definition | | `src/services/document_service.py` | `create_document()` async function | | `scripts/upload_documents.py:203-267` | Airflow REST API trigger pattern | | `docker/docker-compose.airflow.yml:85-104` | Webserver config, env vars, port mapping | | `airflow/dags/document_extraction_dag.py:255-259` | DAG params: document_id, object_key, tenant_id, node_id | ### Technical Decisions - **Flask Blueprint (verified for Airflow 3.0.6)**: Airflow webserver is Flask-based. `flask_blueprints` on `AirflowPlugin` is the native mechanism. Confirmed in Airflow 3.0.6 source — hybrid architecture (Flask webserver + FastAPI API server) means both coexist. - **Airflow DAG trigger API**: Use this import resolution order (Airflow 3.x restructured internals): 1. Try: `from airflow.api.common.trigger_dag import trigger_dag` (Airflow 2.x/early 3.x) 2. Fallback: `from airflow.models.dagrun import DagRun` + `DagRun.generate_run_id()` + `dag.create_dagrun(run_id=..., conf=..., run_type=DagRunType.MANUAL, state=DagRunState.QUEUED, logical_date=timezone.utcnow())` 3. Verify at container startup: run `python -c "from airflow.api.common.trigger_dag import trigger_dag"` in Dockerfile or init script to catch import errors early - **Sync DB operations**: Flask blueprints are synchronous. Use `sqlalchemy.create_engine` with `psycopg2` driver (installed in container). Connection string: `DATABASE_URL` env var with `+asyncpg` replaced by `+psycopg2`. Create engine once at module level, create session per-request, close after use. - **api_key_id = None**: Internal service-to-service uploads don't have an API key. `api_key_id` on `Document` is nullable. - **Source bucket via env var**: `UPLOAD_BUCKET` env var (default: `source-files` per data-isolation-spec Section 4.2). Current code diverges — `ValidateEventOperator.DEFAULT_BUCKET = 'documents'` and `upload_documents.py` defaults to `documents`. The plugin reads `os.getenv('UPLOAD_BUCKET', 'source-files')` and passes it in DAG conf. The `ValidateEventOperator` already reads `conf.get('bucket', self.DEFAULT_BUCKET)` so passing it explicitly works without operator changes. - **No batch upload v1**: Single file per request. Batch can be added later as `POST /api/v1/documents/upload-batch`. - **Full HMAC verification per SPEC-HMAC**: Implement complete HMAC-SHA256 signature verification (Section 3.3) — extract 7 headers, validate formats, check timestamp freshness (±30s configurable), load key ring from env/file, reconstruct canonical string (alphabetical sort, pipe-separated), constant-time signature comparison. This makes the endpoint secure from day one rather than deferring to GAP-017. - **Import from `src`**: Since PYTHONPATH includes `/opt/airflow/src`, import `src.services.file_validator`, `src.services.seaweedfs_service`, `src.db.models` directly. No code duplication needed. ## Implementation Plan ### Tasks > Tasks ordered by dependency. Each task is a discrete, completable unit. #### Task 1: HMAC Auth Module (full verification per SPEC-HMAC) **File:** `airflow/plugins/api/auth.py` (NEW) - [x] Create `airflow/plugins/api/` package with `__init__.py` - [x] Create `auth.py` implementing SPEC-HMAC Section 3.3 verification: **Step 1 — Extract all 7 required headers:** - `X-Tenant-ID` (UUID) — tenant identifier - `X-Node-ID` (UUID) — hierarchy node context - `X-User-ID` (UUID) — authenticated user - `X-Role` (enum: `manager`, `content_specialist`, `reviewer`, `end_user`) - `X-Timestamp` (ISO 8601 UTC, e.g. `2026-03-11T08:30:00Z`) - `X-Key-ID` (string) — key version identifier - `X-Signature` (hex string, 64 chars) — HMAC-SHA256 signature - Validate all present → 401 `{"detail": "Missing header: X-Tenant-ID"}` if any missing **Step 2 — Validate field formats:** - UUID format for `X-Tenant-ID`, `X-Node-ID`, `X-User-ID` via `uuid.UUID(value)` - `X-Role` in `{"manager", "content_specialist", "reviewer", "end_user"}` → 401 `"Invalid role"` - `X-Signature` is 64-char lowercase hex → 401 if malformed **Step 3 — Timestamp freshness check (SPEC-HMAC Section 3.3):** - Parse `X-Timestamp` as ISO 8601 UTC - Compare to `datetime.now(timezone.utc)` - Reject if drift > `settings.hmac_timestamp_tolerance` seconds (default: 30) - → 401 `"Request expired or clock drift too large"` **Step 4 — Key ring loading (SPEC-HMAC Section 5.2):** - Load from `settings.hmac_key_ring` (JSON string, takes priority) or `settings.hmac_key_ring_file` (file path) - Structure: `{"active_key_id": "v2", "keys": {"v1": "...", "v2": "..."}}` - Resolve secret by `X-Key-ID` → 401 `"Unknown key version"` if not found - Cache key ring at module level, reload on file change (optional for v1) **Step 5 — Signature verification (SPEC-HMAC Section 3.1):** - Reconstruct canonical string: sort 5 signed headers alphabetically (`X-Node-ID`, `X-Role`, `X-Tenant-ID`, `X-Timestamp`, `X-User-ID`), join values with `|` - Compute expected: `hmac.new(secret.encode('utf-8'), string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()` - **Constant-time compare**: `hmac.compare_digest(expected, received_signature)` → 401 `"Invalid signature"` on mismatch **Step 6 — Return verified context:** - Return dict: `{"tenant_id": str, "node_id": str, "user_id": str, "role": str}` All config via `src/core/config.py` `Settings` class (auto-loaded from `.env`). #### Task 2: Extend Settings Class **File:** `src/core/config.py` - [x] Add new fields to `Settings`: ```python # Upload endpoint upload_bucket: str = "source-files" max_filename_length: int = 255 # HMAC verification hmac_key_ring: str = "" # JSON string, or use hmac_key_ring_file hmac_key_ring_file: str = "/vault/secrets/hmac-key-ring.json" hmac_timestamp_tolerance: int = 30 ``` These are auto-loaded from `.env` via Pydantic `SettingsConfigDict(env_file=".env")`. #### Task 3: Upload Blueprint — Scaffold + S3 Client **File:** `airflow/plugins/api/upload_blueprint.py` (NEW) - [x] Create Flask Blueprint: `upload_bp = Blueprint('upload', __name__, url_prefix='/api/v1/documents')` - [x] Create module-level S3 client (lazy init): ```python import os, boto3 from botocore.client import Config def get_s3_client(): return boto3.client( 's3', endpoint_url=os.getenv('AWS_ENDPOINT_URL'), config=Config(signature_version='s3v4'), region_name='us-east-1', ) ``` - [x] Import settings: `from src.core.config import settings` - [x] Create module-level sync SQLAlchemy engine (lazy init, F9 fix): ```python from sqlalchemy import create_engine from sqlalchemy.orm import Session _engine = None def get_engine(): global _engine if _engine is None: url = settings.sync_database_url # already converts +asyncpg → +psycopg2 if not url: raise RuntimeError("DATABASE_URL env var is required") _engine = create_engine(url, pool_pre_ping=True, pool_size=3, max_overflow=2) return _engine ``` - [x] Set Flask `MAX_CONTENT_LENGTH` from settings (F10 fix): ```python # In blueprint or app factory: app.config['MAX_CONTENT_LENGTH'] = settings.max_upload_size_mb * 1024 * 1024 ``` This rejects oversized requests at the Flask level **before** buffering the entire body. - [x] Read bucket from `settings.upload_bucket` (default: `source-files`) #### Task 3: Upload Blueprint — POST /upload Route **File:** `airflow/plugins/api/upload_blueprint.py` - [x] Implement `@upload_bp.route('/upload', methods=['POST'])` with this sequence: 1. **Auth**: Call `verify_hmac_headers(request)` → **401** on any failure (per SPEC-HMAC Section 4.2). Extract `tenant_id`, `node_id`, `user_id`, `role` from returned dict 2. **File extraction**: `file = request.files.get('file')` → 400 if missing 3. **Validation (fail-fast)**: - Extension check: `pathlib.Path(file.filename).suffix.lower()` against whitelist → 400 `INVALID_EXTENSION` - Size check: `file.seek(0, 2)` / `file.tell()` → 400 `FILE_TOO_LARGE` if > `settings.max_upload_size_mb * 1024 * 1024`, then `file.seek(0)` - Magic bytes: read first 8 bytes, verify against `PK` / `\xd0\xcf\x11\xe0` / `%PDF-` → 400 `INVALID_FORMAT` - Save to temp file (`tempfile.NamedTemporaryFile`), run `check_password_protected(temp_path)` → 400 `PASSWORD_PROTECTED` 4. **Sanitize filename** (F1 fix): Strip path traversal but preserve CJK/Unicode characters: ```python import os safe_filename = os.path.basename(filename) # strip directory components (handles / and \) safe_filename = safe_filename.replace('\x00', '') # strip null bytes safe_filename = safe_filename.replace('..', '') # strip path traversal sequences safe_filename = safe_filename.strip('. ') # strip leading/trailing dots and spaces if len(safe_filename) > settings.max_filename_length: return 400 FILENAME_TOO_LONG if not safe_filename: return 400 INVALID_FILENAME ``` This preserves Japanese/Chinese filenames (e.g. `報告書.xlsx`) while blocking path traversal. 5. **Generate IDs**: `document_id = uuid4()`, build `s3_key = f"{tenant_id}/{node_id}/{document_id}/{safe_filename}"` 6. **S3 path validation**: `validate_s3_path(tenant_id, node_id, s3_key)` — uses **resolved** path (no `..` possible after sanitization) 7. **S3 upload**: Ensure bucket exists (`head_bucket` / `create_bucket`), `put_object` from temp file 8. **DB insert** (inside try block — commit deferred): ```python session = Session(engine) try: doc = Document(id=document_id, filename=safe_filename, file_path=s3_key, file_size_bytes=size, status='pending', tenant_id=UUID(tenant_id), node_id=UUID(node_id)) session.add(doc) session.flush() # get ID, don't commit yet ``` 9. **DAG trigger**: ```python dag_run = trigger_dag_run(dag_id, conf) # see Task 5 for API session.commit() # only commit after DAG trigger succeeds except Exception: session.rollback() # cleanup: delete S3 object if it was uploaded s3_client.delete_object(Bucket=bucket, Key=s3_key) raise finally: session.close() ``` 10. **Response**: `jsonify({"document_id": str(document_id), "filename": safe_filename, "status": "pending"})`, status 202 11. **Cleanup**: `finally` block deletes temp file (always, regardless of success/failure) - [x] Error handling: - Auth failures: return **401** with `{"detail": "..."}` (F2 fix — matches SPEC-HMAC Section 4.2) - File validation failures: return 400 with `{"error": {"code": "...", "message": "..."}}` - S3 upload failure: return 500, no DB record committed (flush without commit) - DB failure after S3 upload: rollback DB, delete S3 object (F5 fix — no orphaned files) - DAG trigger failure: rollback DB, delete S3 object, return 500 with `{"error": {"code": "DAG_TRIGGER_FAILED", ...}}` #### Task 4: Register Blueprint in Plugin **File:** `airflow/plugins/__init__.py` - [x] Import: `from api.upload_blueprint import upload_bp` (with try/except for ImportError fallback to relative import) - [x] Add `flask_blueprints = [upload_bp]` to `TextIQPlugin` class - [x] Verify: restart Airflow webserver, confirm `GET /api/v1/documents/upload` returns 405 (method not allowed = route exists) #### Task 5: Fix ValidateEventOperator Extension Map (F6) **File:** `airflow/plugins/operators/validate_event.py` - [x] Add missing extensions to `EXTENSION_TO_CONTENT_TYPE` map: ```python 'ppt': 'application/vnd.ms-powerpoint', 'pptm': 'application/vnd.ms-powerpoint.presentation.macroEnabled.12', ``` - [x] Without this fix, `.ppt`/`.pptm` files uploaded via the plugin will fail at the validate_event task #### Task 6: Environment Config **Files:** `.env.example`, `.env`, `docker/docker-compose.airflow.yml` - [x] Add to `.env.example` (and `.env` for local dev): ```bash # =================== # Upload Endpoint Configuration # =================== # S3 bucket for uploaded source files UPLOAD_BUCKET=source-files # HMAC key ring for request signing (JSON string) # Generate production keys with: python -c "import secrets; print(secrets.token_urlsafe(32))" HMAC_KEY_RING={"active_key_id":"v1","keys":{"v1":"dev-secret-key-min-32-chars-for-hmac-sha256"}} # Max filename length (must match DB column constraint) MAX_FILENAME_LENGTH=255 # Max allowed clock drift for HMAC timestamp verification (seconds) HMAC_TIMESTAMP_TOLERANCE=30 ``` Note: `MAX_UPLOAD_SIZE_MB=50` already exists in `.env.example` (line 91). - [x] Add env var passthrough in `docker/docker-compose.airflow.yml` `x-airflow-env` section: ```yaml UPLOAD_BUCKET: ${UPLOAD_BUCKET:-source-files} HMAC_KEY_RING: ${HMAC_KEY_RING} MAX_FILENAME_LENGTH: ${MAX_FILENAME_LENGTH:-255} HMAC_TIMESTAMP_TOLERANCE: ${HMAC_TIMESTAMP_TOLERANCE:-30} ``` Docker-compose reads from `.env` automatically — no hardcoded values in the compose file. - [x] For production: use `HMAC_KEY_RING_FILE` pointing to Vault-managed secret instead of inline JSON - [x] Verify `psycopg2-binary` is available in container (check `Dockerfile.airflow` or pip list) - [x] Verify `olefile` is available (needed for legacy password detection in `.doc`/`.xls`/`.ppt`) ### Acceptance Criteria **AC-1: Endpoint responds on Airflow webserver** - Given the Airflow webserver is running - When `POST http://localhost:8081/api/v1/documents/upload` is called with a valid file and headers - Then a 202 response with `{document_id, filename, status: "pending"}` is returned **AC-2: Full HMAC verification (per SPEC-HMAC Section 3.3)** - Given a request missing any of the 7 headers (`X-Tenant-ID`, `X-Node-ID`, `X-User-ID`, `X-Role`, `X-Timestamp`, `X-Key-ID`, `X-Signature`) - When the upload endpoint is called - Then a 401 response with `"Missing header: {name}"` is returned - And invalid UUID format returns 401 `"Invalid UUID: {name}"` - And invalid role returns 401 `"Invalid role"` - And expired timestamp (>30s drift) returns 401 `"Request expired or clock drift too large"` - And unknown `X-Key-ID` returns 401 `"Unknown key version"` - And tampered signature returns 401 `"Invalid signature"` - And SPEC-HMAC test vector 1 passes verification successfully **AC-3: File validation rejects invalid uploads** - Given a file with invalid extension (e.g., `.txt`) - When the upload endpoint is called - Then a 400 with `INVALID_EXTENSION` is returned - And similarly for files >50MB (`FILE_TOO_LARGE`) and wrong magic bytes (`INVALID_FORMAT`) **AC-3b: Password-protected files are rejected** - Given a password-protected .xlsx file (OLE2 container instead of ZIP) - When the upload endpoint is called - Then a 400 with `PASSWORD_PROTECTED` error code is returned - And the temp file is cleaned up **AC-4: File is written to SeaweedFS with tenant-scoped key in correct bucket** - Given a valid upload with `tenant_id=aaa` and `node_id=bbb` - When the file is uploaded - Then the S3 key is `aaa/bbb/{document_id}/{filename}` in the `UPLOAD_BUCKET` bucket (default: `source-files`) - And the bucket is auto-created if it doesn't exist **AC-5: Document record created in PostgreSQL** - Given a valid upload - When the endpoint completes - Then a `documents` row exists with `status='pending'`, correct `tenant_id`, `node_id`, `filename`, `file_path`, `file_size_bytes` **AC-6: DAG is triggered with correct conf** - Given a valid upload - When the endpoint completes - Then a `document_extraction_dag` DAG run exists with `conf` containing `document_id`, `bucket: "source-files"`, `object_key`, `tenant_id`, `node_id` **AC-7: S3 path validation blocks cross-tenant keys** - Given `tenant_id=aaa` and `node_id=bbb` - When the generated S3 key does not match the `aaa/bbb/` prefix (should never happen in normal flow) - Then the upload is rejected **AC-8: Cleanup on failure — no orphaned data** - Given a valid file but S3 upload fails - When the error is caught - Then the `documents` DB record is not committed (flush without commit pattern) - And given S3 upload succeeds but DB commit fails - Then the S3 object is deleted (no orphaned files) - And given DAG trigger fails after S3 upload - Then both DB record is rolled back and S3 object is deleted **AC-9: Filename sanitization prevents path traversal (F1)** - Given an uploaded file with filename `../../etc/passwd` - When the upload endpoint processes it - Then the S3 key uses sanitized filename `etcpasswd` (path components and traversal stripped) - And the S3 key still matches `{tenant_id}/{node_id}/{document_id}/` prefix - And filenames with directory separators (`/`, `\`) are stripped **AC-10: HMAC auth returns 401 not 400 (F2)** - Given any HMAC verification failure (missing header, invalid UUID, expired timestamp, bad signature) - When the upload endpoint is called - Then HTTP 401 is returned (not 400) - And error messages match SPEC-HMAC Section 4.2 format ## Additional Context ### Dependencies - **data-isolation-spec Section 6.4**: This endpoint is the implementation of the "DE receives raw file uploads" target flow - **SPEC-HMAC (HMAC Context Headers)**: This endpoint implements SPEC-HMAC Section 3.3 receiver-side verification. BFF senders (APA/UPA) must implement Section 3.2 signing before calling this endpoint. - **GAP-038 (APA write-path migration)**: After this endpoint exists, APA can be refactored to proxy uploads here instead of writing to SeaweedFS directly. APA must add `sign_context_headers()` to outgoing calls. - **Airflow 3.0.6 Flask Blueprint**: Verified supported — `flask_blueprints` attribute on `AirflowPlugin` works in 3.0.6 ### Testing Strategy **Unit Tests:** - `verify_hmac_headers()`: all 7 headers valid + correct signature → returns context dict, missing header → 401, invalid UUID → 401, invalid role → 401, expired timestamp → 401, bad key ID → 401, tampered signature → 401 (constant-time compare) - SPEC-HMAC test vectors: vector 1 (content_specialist) passes, vector 2 (end_user) passes, vector 3 (tampered tenant) fails with "Invalid signature" - File validation: each of 10 extensions accepted, `.txt`/`.csv` rejected, 49MB pass, 51MB fail, correct magic bytes pass, wrong magic bytes fail - Password detection: password-protected .xlsx (OLE2 magic) → `PASSWORD_PROTECTED`, normal .xlsx (PK magic) → pass - S3 key generation: verify format `{tenant_id}/{node_id}/{document_id}/{filename}`, verify `validate_s3_path` passes **Integration Tests (Docker environment):** - Upload valid .xlsx → verify S3 `head_object` returns 200 at correct key in `source-files` bucket - Upload valid .xlsx → verify `documents` DB row with `status='pending'`, correct `tenant_id`, `node_id`, `file_path`, `file_size_bytes` - Upload valid .xlsx → verify `document_extraction_dag` DAG run exists with correct `conf` - Upload without headers → verify 401 with `"Missing header: X-Tenant-ID"` - Upload with invalid role → verify 401 with `"Invalid role"` - Upload with tampered X-Tenant-ID (SPEC-HMAC test vector 3) → verify 401 `"Invalid signature"` - Upload with expired timestamp (>30s old) → verify 401 `"Request expired..."` - Upload with correct SPEC-HMAC test vector 1 headers → verify 202 success - Upload .txt file → verify 400 with `INVALID_EXTENSION` - Upload password-protected .xlsx → verify 400 with `PASSWORD_PROTECTED` - S3 failure (e.g., wrong endpoint) → verify 500, no orphaned DB record **End-to-End:** ```bash # Sample upload request (compute X-Signature from your HMAC_KEY_RING secret) curl -X POST http://localhost:8081/api/v1/documents/upload \ -H "X-Tenant-ID: 11111111-1111-1111-1111-111111111111" \ -H "X-Node-ID: 22222222-2222-2222-2222-222222222222" \ -H "X-User-ID: 33333333-3333-3333-3333-333333333333" \ -H "X-Role: content_specialist" \ -H "X-Timestamp: $(date -u +%Y-%m-%dT%H:%M:%SZ)" \ -H "X-Key-ID: v1" \ -H "X-Signature: " \ -F "file=@test.xlsx" # → 202 {"document_id": "...", "filename": "test.xlsx", "status": "pending"} # → DAG runs → document processed → query returns results ``` ### Notes - **Flask Blueprint confirmed supported** in Airflow 3.0.6 — hybrid architecture (Flask webserver + FastAPI API server) means blueprints attach to the Flask side. - **Header count clarification (F4)**: 7 total headers in the HMAC contract. 5 are signed (X-Node-ID, X-Role, X-Tenant-ID, X-Timestamp, X-User-ID — these form the canonical string). 2 are transport metadata (X-Key-ID selects the key, X-Signature carries the signature). All 7 are required on every request. - **Concurrency ceiling (F10)**: The endpoint is synchronous (Flask). Each upload holds a DB connection (pool size=3, overflow=2) and blocks on S3 upload. Max ~5 concurrent uploads before DB pool exhaustion. This is acceptable for the expected low-volume BFF-proxy usage pattern. If higher throughput is needed, consider an async endpoint or dedicated upload service. - **Size limit mismatch (F7)**: Upload endpoint enforces 50MB (Flask `MAX_CONTENT_LENGTH` + file validator). `ValidateEventOperator` allows 100MB. The upload endpoint is stricter — no risk. Consider aligning the operator to 50MB in a follow-up. - `api_key_id` is set to None for internal uploads. If API key auth is needed later, add as optional `X-Api-Key` header. - The `source-files` bucket must exist in SeaweedFS before the first upload. The endpoint auto-creates it if missing. - The upload script (`scripts/upload_documents.py`) uses `documents` as default bucket — this endpoint uses `source-files` per data-isolation-spec Section 4.2. These are different buckets. - **Filename sanitization (F1)**: Directory separators (`/`, `\`), null bytes, and `..` traversal sequences are stripped. Original Unicode characters (CJK, etc.) are preserved — `報告書.xlsx` stays as-is. Max 255 chars.