changement vers google drive storage pour les contrats.

Tentative d'obtenir un token de connection échoue

_add_drive_test.py

@@ -0,0 +1,40 @@
+"""Add Drive signed upload test."""
+import pathlib
+
+p = pathlib.Path('tests/test_contract_uploads.py')
+t = p.read_bytes()
+
+drive_test = (
+    b"\n    def test_signed_upload_with_google_drive(\n"
+    b"        self, app, client, as_role, make_user, contract_for, monkeypatch\n"
+    b"    ):\n"
+    b'        """Signing with Drive backend stores gdrive://<id> in signed_file_path."""\n'
+    b"        monkeypatch.setenv('DOCUMENT_STORAGE_BACKEND', 'google_drive')\n"
+    b"        monkeypatch.setattr(\n"
+    b"            'app.storage.upload_google_drive_file',\n"
+    b"            lambda **_kwargs: 'drive-signed-456',\n"
+    b"        )\n"
+    b"        player_id = as_role('player')\n"
+    b"        admin_id = make_user('admin')\n"
+    b"        contract_id, directory = contract_for(player_id, admin_id)\n\n"
+    b"        client.post(\n"
+    b"            f'/users/contracts/{contract_id}/upload_signed',\n"
+    b"            data={'signed_file': (io.BytesIO(_PDF_BYTES), 'signed.pdf')},\n"
+    b"            content_type='multipart/form-data',\n"
+    b"            follow_redirects=True,\n"
+    b"        )\n\n"
+    b"        with app.app_context():\n"
+    b"            contract = db.session.get(Contract, contract_id)\n"
+    b"            assert contract.status == 'signed'\n"
+    b"            assert contract.signed_file_path == 'gdrive://drive-signed-456'\n\n"
+)
+
+# Add before 'class TestTheHelper:'
+marker = b'\nclass TestTheHelper:'
+pos = t.find(marker)
+if pos >= 0:
+    t = t[:pos] + drive_test + t[pos:]
+    p.write_bytes(t)
+    print('Added Drive signed upload test')
+else:
+    print('Could not find marker')

_add_tests.py

@@ -0,0 +1,48 @@
+"""Add new tests to test_contract_uploads.py."""
+import pathlib
+
+p = pathlib.Path('tests/test_contract_uploads.py')
+t = p.read_bytes()
+
+# 1. Header-only test in TestSignedUpload (before 'class TestTheHelper:')
+marker = b'\n\nclass TestTheHelper:'
+test = (
+    b"\n    def test_a_header_only_pdf_is_refused(\n"
+    b"        self, app, client, as_role, make_user, contract_for\n"
+    b"    ):\n"
+    b'        """A file with a %PDF- header but no real PDF structure is rejected."""\n'
+    b"        player_id = as_role('player')\n"
+    b"        admin_id = make_user('admin')\n"
+    b"        contract_id, directory = contract_for(player_id, admin_id)\n\n"
+    b"        client.post(\n"
+    b"            f'/users/contracts/{contract_id}/upload_signed',\n"
+    b"            data={'signed_file': (io.BytesIO(PDF_HEADER_ONLY_BYTES), 'signed.pdf')},\n"
+    b"            content_type='multipart/form-data',\n"
+    b"            follow_redirects=True,\n"
+    b"        )\n\n"
+    b"        with app.app_context():\n"
+    b"            contract = db.session.get(Contract, contract_id)\n"
+    b"            assert contract.status != 'signed'\n"
+    b"            assert contract.signed_file_path is None\n\n"
+)
+pos = t.find(marker)
+if pos >= 0:
+    t = t[:pos] + test + t[pos:]
+
+# 2. Header-only helper test at end of TestTheHelper
+trailing = t.rstrip(b'\n')
+idx = t.find(trailing) + len(trailing)
+helper = (
+    b"\n    def test_a_header_only_file_is_refused(self, app):\n"
+    b'        """A stream starting with %PDF- but holding non-PDF data is\n'
+    b"        rejected (the header check is not enough).\"\"\"\n"
+    b"        from werkzeug.datastructures import FileStorage\n\n"
+    b"        from app.routes.users import ALLOWED_CONTRACT_EXTENSIONS, pdf_upload_error\n\n"
+    b"        upload = FileStorage(stream=io.BytesIO(PDF_HEADER_ONLY_BYTES), filename='clever.pdf')\n"
+    b"        with app.test_request_context('/'):\n"
+    b"            assert pdf_upload_error(upload, ALLOWED_CONTRACT_EXTENSIONS)\n"
+)
+t = t[:idx] + helper + t[idx:]
+
+p.write_bytes(t)
+print('Added new tests')

app/google_drive.py

@@ -22,7 +22,9 @@ class GoogleDriveStorageError(RuntimeError):
 def _setting(name):
     value = os.getenv(name)
     if not value:
-        raise GoogleDriveStorageError(f'{name} must be configured for Google Drive document storage.')
+        raise GoogleDriveStorageError(
+            f'{name} must be configured for Google Drive document storage.'
+        )
     return value
 
 
@@ -47,6 +49,43 @@ def _drive_service():
     return build('drive', 'v3', credentials=credentials, cache_discovery=False)
 
 
+def list_folder_files():
+    """Return (file_id, name) tuples for every file in the configured folder.
+
+    Used by the backup script to enumerate Drive-stored contracts. Returns
+    only files directly in the folder (not nested inside sub-folders).
+
+    Returns:
+        list[tuple[str, str]]: (file_id, name) for each file found.
+    """
+    try:
+        folder_id = _setting(GOOGLE_DRIVE_FOLDER_ID_ENV)
+        page_token = None
+        results = []
+        while True:
+            response = (
+                _drive_service()
+                .files()
+                .list(
+                    q=f"'{folder_id}' in parents and trashed=false",
+                    fields='nextPageToken, files(id, name)',
+                    pageToken=page_token,
+                    pageSize=1000,
+                )
+                .execute()
+            )
+            for f in response.get('files', []):
+                results.append((f['id'], f['name']))
+            page_token = response.get('nextPageToken')
+            if not page_token:
+                break
+        return results
+    except GoogleDriveStorageError:
+        raise
+    except Exception as exc:
+        raise GoogleDriveStorageError('Google Drive could not list folder contents.') from exc
+
+
 def upload_file(*, stream, filename, mimetype):
     """Upload a PDF to the configured owner folder and return its Drive ID."""
     try:

app/routes/users/_shared.py

@@ -39,6 +39,10 @@ def pdf_upload_error(file, allowed_extensions):
     could put an arbitrary file on the server under a name the application
     later hands back for download (SEC-021).
 
+    The check now goes beyond the header: it parses the upload with pypdf
+    and requires at least one parseable page, so a file with ``%PDF-``
+    prepended or a renamed non-PDF is rejected.
+
     Args:
         file: The uploaded FileStorage, or None.
         allowed_extensions: Extensions to accept, lowercase and without dot.
@@ -53,11 +57,27 @@ def pdf_upload_error(file, allowed_extensions):
     if not (stem and dot) or extension.lower() not in allowed_extensions:
         return _('Only PDF files are allowed for contracts.')
 
+    # Fast-fail: the PDF signature must appear at the start of the stream.
     head = file.stream.read(len(PDF_SIGNATURE))
     file.stream.seek(0)
     if head != PDF_SIGNATURE:
         return _('That file is not a PDF, whatever its name says.')
 
+    # Full-content validation: parse with pypdf and require at least one
+    # page.  A file with a forged header but no real PDF structure (e.g. a
+    # renamed executable with ``%PDF-`` prepended) will fail here.
+    try:
+        from pypdf import PdfReader
+        from pypdf.errors import PyPdfError
+
+        reader = PdfReader(file.stream, strict=False)
+        if len(reader.pages) == 0:
+            return _('That file is not a PDF, whatever its name says.')
+    except (PyPdfError, ValueError, TypeError):
+        return _('That file is not a PDF, whatever its name says.')
+    finally:
+        file.stream.seek(0)
+
     return None
 
 

app/routes/users/contracts.py

@@ -135,7 +135,9 @@ def upload_contract():
         try:
             stored_path = store_uploaded_document(file, relative_path)
         except GoogleDriveStorageError:
-            flash(_('Contract storage is temporarily unavailable. Please try again later.'), 'danger')
+            flash(
+                _('Contract storage is temporarily unavailable. Please try again later.'), 'danger'
+            )
             return redirect(url_for('users.upload_contract'))
 
         contract = Contract(

app/storage.py

@@ -34,8 +34,14 @@ import os
 
 from app.google_drive import (
     GoogleDriveStorageError,
+)
+from app.google_drive import (
     delete_file as delete_google_drive_file,
+)
+from app.google_drive import (
     download_file as download_google_drive_file,
+)
+from app.google_drive import (
     upload_file as upload_google_drive_file,
 )
 
@@ -112,7 +118,7 @@ def _rooted(env_name, default_name):
 
 def document_storage_backend():
     """Return the configured backend for new document uploads."""
-    backend = os.getenv(DOCUMENT_STORAGE_BACKEND_ENV, 'google_drive').strip().lower()
+    backend = os.getenv(DOCUMENT_STORAGE_BACKEND_ENV, 'local').strip().lower()
     if backend not in {'google_drive', 'local'}:
         raise ValueError(
             f'{DOCUMENT_STORAGE_BACKEND_ENV} must be "google_drive" or "local", not {backend!r}'

app/supporting_scripts/backup.py

@@ -42,7 +42,11 @@ from urllib.parse import unquote, urlparse
 # to be — see DOCUMENTS_DIR below.
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
 
-from app.storage import backups_root, documents_root  # noqa: E402 — needs the path above
+from app.storage import (  # noqa: E402 — needs the path above
+    backups_root,
+    document_storage_backend,
+    documents_root,
+)
 
 # Configuration
 BACKUP_DIR = backups_root()
@@ -292,6 +296,75 @@ def backup_documents():
     return zip_path
 
 
+def backup_drive_documents():
+    """Download every file from the configured Google Drive folder and archive it.
+
+    Called **in addition to** ``backup_documents()`` when the storage backend is
+    ``google_drive``.  When the backend is ``local`` this function silently
+    skips — local backups already handled by ``backup_documents()``.
+
+    Returns:
+        str | None: Path to the created archive, or None when skipped.
+
+    Raises:
+        BackupError: If Drive is configured but listing or downloading fails.
+    """
+    if document_storage_backend() != 'google_drive':
+        print('[SKIP] Google Drive is not configured; local documents only.')
+        return None
+
+    try:
+        from app.google_drive import (
+            GoogleDriveStorageError,
+            list_folder_files,
+        )
+        from app.google_drive import (
+            download_file as download_drive_file,
+        )
+    except ImportError:
+        raise BackupError(
+            'Google Drive dependencies are not installed. Install requirements.txt again.'
+        ) from None
+
+    print('[INFO] Listing Google Drive contract folder...')
+    try:
+        files = list_folder_files()
+    except (GoogleDriveStorageError, Exception) as exc:  # noqa: BLE001 — boundary
+        raise BackupError(f'Google Drive listing failed: {exc}') from exc
+
+    if not files:
+        print('[INFO] Google Drive folder is empty; no Drive documents to back up.')
+        return None
+
+    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+    archive_basename = os.path.join(BACKUP_DIR, f'documents_backup_drive_{timestamp}')
+
+    download = {}
+    for file_id, name in files:
+        print(f'  Downloading {name} ({file_id})...')
+        try:
+            download[name] = download_drive_file(file_id)
+        except (GoogleDriveStorageError, Exception) as exc:  # noqa: BLE001 — boundary
+            raise BackupError(f'Google Drive download failed for {name}: {exc}') from exc
+
+    try:
+        import zipfile
+
+        zip_path = f'{archive_basename}.zip'
+        with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf:
+            for name, data in download.items():
+                zf.writestr(name, data)
+    except Exception as exc:  # noqa: BLE001 — boundary
+        raise BackupError(f'Drive document archive failed: {exc}') from exc
+
+    if not os.path.exists(zip_path) or os.path.getsize(zip_path) == 0:
+        raise BackupError('Drive document archiver reported success but produced an empty file.')
+
+    size_mb = os.path.getsize(zip_path) / (1024 * 1024)
+    print(f'[OK] Drive documents backed up to: {zip_path} ({size_mb:.1f} MB)')
+    return zip_path
+
+
 def cleanup_old_backups():
     """Remove backup files older than BACKUP_RETENTION_DAYS."""
     if not os.path.exists(BACKUP_DIR):
@@ -365,10 +438,16 @@ def main(argv=None):
     except BackupError as exc:
         documents_ok = False
         print(f'[ERROR] {exc}')
+    drive_ok = True
+    try:
+        backup_drive_documents()
+    except BackupError as exc:
+        drive_ok = False
+        print(f'[ERROR] {exc}')
     cleanup_old_backups()
 
     print()
-    if verified and documents_ok:
+    if verified and documents_ok and drive_ok:
         print('=== Backup completed successfully ===')
         return 0
 

app/supporting_scripts/get_refresh_token.py

@@ -0,0 +1,121 @@
+"""Obtain a Google Drive refresh token for the application.
+
+Run this once from a development machine that has a browser. It opens a local
+OAuth redirect server, asks you to log in, and prints the refresh token.
+
+Usage:
+    python app/supporting_scripts/get_refresh_token.py
+
+Requires:
+    pip install google-auth-oauthlib
+
+Before running:
+    1. Create a Google Cloud project and enable the Drive API.
+    2. Create a **Desktop** OAuth 2.0 client ID.
+    3. Add your email as a test user in the OAuth consent screen.
+    4. Run this script and paste the values when prompted.
+"""
+
+import os
+import sys
+
+
+def _prompt(label, env_name, default=""):
+    """Ask the user for a value, accepting an env-var fallback or a default."""
+    candidate = os.getenv(env_name) or default
+    hint = f" [{candidate}]" if candidate else ""
+    raw = input(f"{label}{hint}: ").strip()
+    if raw:
+        return raw
+    if candidate:
+        return candidate
+    return default
+
+
+def _validate_client_id(client_id):
+    """Warn if the client ID doesn't look like a real OAuth client ID."""
+    if client_id == "YOUR_CLIENT_ID.apps.googleusercontent.com":
+        print("[ERROR] You must provide a real client ID from the Google Cloud console.")
+        return False
+    if not client_id.endswith(".apps.googleusercontent.com"):
+        print(
+            f"[WARNING] Client ID should end with '.apps.googleusercontent.com'.\n"
+            f"          Got: {client_id!r}\n"
+            f"          If this is a placeholder, the OAuth request will fail.\n"
+        )
+    return True
+
+
+def _validate_client_secret(secret):
+    if secret == "YOUR_CLIENT_SECRET":
+        print("[ERROR] You must provide a real client secret from the Google Cloud console.")
+        return False
+    # The GOCSPX- prefix is now used by every client type, so it no longer
+    # tells us anything about whether the client was created as a Desktop app.
+    return True
+
+
+def main():
+    print("=" * 60)
+    print("Google Drive Refresh Token — Setup")
+    print("=" * 60)
+    print()
+
+    # When both env vars are set, skip prompts entirely for headless runs.
+    env_id = os.getenv("GOOGLE_DRIVE_CLIENT_ID")
+    env_secret = os.getenv("GOOGLE_DRIVE_CLIENT_SECRET")
+    if env_id and env_secret:
+        client_id = env_id
+        client_secret = env_secret
+        print(f"Using Client ID from environment: {client_id[:30]}...")
+    else:
+        print("Enter your Google Cloud OAuth credentials.")
+        print("You can also set GOOGLE_DRIVE_CLIENT_ID and GOOGLE_DRIVE_CLIENT_SECRET")
+        print("environment variables and re-run to skip the prompts.")
+        print()
+
+        client_id = _prompt("Client ID", "GOOGLE_DRIVE_CLIENT_ID")
+        client_secret = _prompt("Client Secret", "GOOGLE_DRIVE_CLIENT_SECRET")
+
+        if not _validate_client_id(client_id) or not _validate_client_secret(client_secret):
+            sys.exit(1)
+
+    print()
+    print("Opening browser to authenticate with Google...")
+    print("(If it doesn't open, copy the URL below into your browser.)")
+    print()
+
+    from google_auth_oauthlib.flow import InstalledAppFlow
+
+    SCOPE = "https://www.googleapis.com/auth/drive.file"
+
+    client_config = {
+        "installed": {
+            "client_id": client_id,
+            "client_secret": client_secret,
+            "redirect_uris": ["http://localhost"],
+            "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+            "token_uri": "https://oauth2.googleapis.com/token",
+        }
+    }
+
+    flow = InstalledAppFlow.from_client_config(client_config, scopes=[SCOPE])
+    credentials = flow.run_local_server(host="localhost", port=8080, open_browser=True)
+
+    print()
+    print("=" * 60)
+    print("SUCCESS — copy these values into your .env file:")
+    print("=" * 60)
+    print()
+    print("DOCUMENT_STORAGE_BACKEND=google_drive")
+    print(f"GOOGLE_DRIVE_CLIENT_ID={credentials.client_id}")
+    print(f"GOOGLE_DRIVE_CLIENT_SECRET={client_secret}")
+    print(f"GOOGLE_DRIVE_REFRESH_TOKEN={credentials.refresh_token}")
+    print("GOOGLE_DRIVE_FOLDER_ID=<get this from the Drive folder URL>")
+    print()
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

app/templates/pages/contracts.html

@@ -94,8 +94,8 @@
                 <input type="hidden" name="csrf_token" value="{{ csrf_token() }}"/>
                 <div class="form-group">
                     <label for="signed_file">{{ _('Select Signed Contract File') }}</label>
-                    <input type="file" id="signed_file" name="signed_file" accept=".pdf,.doc,.docx,.jpg,.jpeg,.png" required>
+                    <input type="file" id="signed_file" name="signed_file" accept=".pdf" required>
-                    <small class="form-text text-muted">{{ _('Accepted formats: PDF, DOC, DOCX, JPG, PNG') }}</small>
+                    <small class="form-text text-muted">{{ _('Accepted format: PDF') }}</small>
                 </div>
                 <div class="form-actions">
                     <button type="button" class="btn btn-secondary" data-action="hide-upload-signed">{{ _('Cancel') }}</button>

app/templates/pages/upload_contract.html

@@ -24,8 +24,8 @@
             
             <div class="form-group">
                 <label for="contract_file">{{ _('Contract File') }}</label>
-                <input type="file" id="contract_file" name="contract_file" accept=".pdf,.doc,.docx,.jpg,.jpeg,.png" required>
+                <input type="file" id="contract_file" name="contract_file" accept=".pdf" required>
-                <small class="form-text text-muted">{{ _('Accepted formats: PDF, DOC, DOCX, JPG, PNG') }}</small>
+                <small class="form-text text-muted">{{ _('Accepted format: PDF') }}</small>
             </div>
             
             <div class="form-group">

docs/database-restore.md

@@ -15,6 +15,7 @@ incident.
 |---|---|---|
 | `db_backup_<timestamp>.dump` | Full PostgreSQL dump, custom format | Every account, tryout, evaluation, note and contract record |
 | `documents_backup_<timestamp>.zip` | `documents/` directory | The contract **files** themselves — they exist only on disk, the database stores paths |
+| `documents_backup_drive_<timestamp>.zip` | Google Drive contract folder | Drive-stored contracts downloaded and archived when `DOCUMENT_STORAGE_BACKEND=google_drive` |
 
 Losing either one alone loses data. A database restore without the document
 archive leaves contract rows pointing at files that no longer exist.

docs/security-checklist.md

@@ -82,8 +82,9 @@ Run through this checklist before deploying to production.
 
 - [ ] Upload size limited to 16MB (`MAX_CONTENT_LENGTH`)
 - [ ] File extensions restricted to PDF only
+- [ ] File content validated by parsing with pypdf (`PdfReader`) — header (`%PDF-`) + at least one parseable page
 - [ ] Files stored with UUID filenames (not original names)
-- [ ] Upload directory outside web root
+- [ ] Upload directory outside web root (or Google Drive in production)
 
 ## Phase 13 — Logging & Monitoring
 

requirements.txt

@@ -26,6 +26,7 @@ frozenlist==1.8.0
 greenlet==3.5.4
 google-api-python-client==2.198.0
 google-auth==2.56.3
+pypdf==6.16.2
 idna==3.18
 itsdangerous==2.2.0
 Jinja2==3.1.6

tests/test_contract_uploads.py

@@ -14,13 +14,25 @@ import io
 import os
 
 import pytest
+from pypdf import PdfWriter as _Pw
 
 from app.extensions import db
 from app.models import Contract
 
-PDF_BYTES = b'%PDF-1.7\n1 0 obj\n<<>>\nendobj\ntrailer\n%%EOF\n'
+
+def _valid_pdf_bytes():
+    w = _Pw()
+    w.add_blank_page(612, 792)
+    buf = io.BytesIO()
+    w.write(buf)
+    return buf.getvalue()
+
+
+_PDF_BYTES = _valid_pdf_bytes()
 NOT_PDF_BYTES = b'MZ\x90\x00\x03\x00\x00\x00'  # a Windows executable header
 
+PDF_HEADER_ONLY_BYTES = b'%PDF-1.7\n' + b'MZ' * 100  # PDF header over non-PDF content
+
 
 @pytest.fixture
 def contract_for(app, tmp_path):
@@ -29,7 +41,7 @@ def contract_for(app, tmp_path):
     def _make(player_id, uploader_id):
         stored = 'deadbeef.pdf'
         path = tmp_path / stored
-        path.write_bytes(PDF_BYTES)
+        path.write_bytes(_PDF_BYTES)
         with app.app_context():
             contract = Contract(
                 player_id=player_id,
@@ -89,7 +101,7 @@ class TestSignedUpload:
 
         client.post(
             f'/users/contracts/{contract_id}/upload_signed',
-            data={'signed_file': (io.BytesIO(PDF_BYTES), 'signed.pdf')},
+            data={'signed_file': (io.BytesIO(_PDF_BYTES), 'signed.pdf')},
             content_type='multipart/form-data',
             follow_redirects=True,
         )
@@ -110,7 +122,7 @@ class TestSignedUpload:
 
         client.post(
             f'/users/contracts/{contract_id}/upload_signed',
-            data={'signed_file': (io.BytesIO(PDF_BYTES), 'signed.pdf')},
+            data={'signed_file': (io.BytesIO(_PDF_BYTES), 'signed.pdf')},
             content_type='multipart/form-data',
             follow_redirects=True,
         )
@@ -118,6 +130,31 @@ class TestSignedUpload:
         with app.app_context():
             assert db.session.get(Contract, contract_id).status != 'signed'
 
+    def test_signed_upload_with_google_drive(
+        self, app, client, as_role, make_user, contract_for, monkeypatch
+    ):
+        """Signing with Drive backend stores gdrive://<id> in signed_file_path."""
+        monkeypatch.setenv('DOCUMENT_STORAGE_BACKEND', 'google_drive')
+        monkeypatch.setattr(
+            'app.storage.upload_google_drive_file',
+            lambda **_kwargs: 'drive-signed-456',
+        )
+        player_id = as_role('player')
+        admin_id = make_user('admin')
+        contract_id, directory = contract_for(player_id, admin_id)
+
+        client.post(
+            f'/users/contracts/{contract_id}/upload_signed',
+            data={'signed_file': (io.BytesIO(_PDF_BYTES), 'signed.pdf')},
+            content_type='multipart/form-data',
+            follow_redirects=True,
+        )
+
+        with app.app_context():
+            contract = db.session.get(Contract, contract_id)
+            assert contract.status == 'signed'
+            assert contract.signed_file_path == 'gdrive://drive-signed-456'
+
 
 class TestTheHelper:
     def test_it_reports_a_missing_file(self, app):
@@ -133,16 +170,27 @@ class TestTheHelper:
 
         from app.routes.users import ALLOWED_CONTRACT_EXTENSIONS, pdf_upload_error
 
-        upload = FileStorage(stream=io.BytesIO(PDF_BYTES), filename='c.pdf')
+        upload = FileStorage(stream=io.BytesIO(_PDF_BYTES), filename='c.pdf')
         with app.test_request_context('/'):
             assert pdf_upload_error(upload, ALLOWED_CONTRACT_EXTENSIONS) is None
-        assert upload.stream.read() == PDF_BYTES
+        assert upload.stream.read() == _PDF_BYTES
 
     def test_a_name_without_a_dot_is_refused(self, app):
         from werkzeug.datastructures import FileStorage
 
         from app.routes.users import ALLOWED_CONTRACT_EXTENSIONS, pdf_upload_error
 
-        upload = FileStorage(stream=io.BytesIO(PDF_BYTES), filename='pdf')
+        upload = FileStorage(stream=io.BytesIO(_PDF_BYTES), filename='pdf')
+        with app.test_request_context('/'):
+            assert pdf_upload_error(upload, ALLOWED_CONTRACT_EXTENSIONS)
+
+    def test_a_header_only_file_is_refused(self, app):
+        """A stream starting with %PDF- but holding non-PDF data is
+        rejected (the header check is not enough)."""
+        from werkzeug.datastructures import FileStorage
+
+        from app.routes.users import ALLOWED_CONTRACT_EXTENSIONS, pdf_upload_error
+
+        upload = FileStorage(stream=io.BytesIO(PDF_HEADER_ONLY_BYTES), filename='clever.pdf')
         with app.test_request_context('/'):
             assert pdf_upload_error(upload, ALLOWED_CONTRACT_EXTENSIONS)

tests/test_storage.py

@@ -28,6 +28,20 @@ from app.storage import (
 )
 
 
+def _valid_pdf():
+    """A real one-page PDF that passes pypdf content validation."""
+    import io
+
+    from pypdf import PdfWriter
+
+    w = PdfWriter()
+    w.add_blank_page(612, 792)
+    buf = io.BytesIO()
+    w.write(buf)
+    buf.seek(0)
+    return buf
+
+
 class TestDocumentsRoot:
     def test_the_environment_wins(self, tmp_path, monkeypatch):
         monkeypatch.setenv('DOCUMENTS_ROOT', str(tmp_path / 'elsewhere'))
@@ -165,10 +179,8 @@ class TestThroughTheUploadRoute:
 
 
 def _pdf():
-    """The smallest thing pdf_upload_error accepts."""
+    """The smallest valid one-page PDF that passes pypdf content validation."""
-    import io
+    return _valid_pdf()
-
-    return io.BytesIO(b'%PDF-1.4\n%%EOF\n')
 
 
 class TestDeletingAnAccountTakesItsDocuments: