team-tryouts/tests/test_contract_uploads.py

"""What may be written to the contracts directory — SEC-021.

upload_signed_contract accepted any file with a non-empty name.
ALLOWED_SIGNED_EXTENSIONS was declared next to it and never read. The file
landed on disk under a name the application later hands back through
download_signed_contract, so whatever a player uploaded is what a manager
opens.

The name alone was never enough either: `payload.pdf` says nothing about
the bytes. Both routes now check the signature as well.
"""

import io
import os

import pytest
from pypdf import PdfWriter as _Pw

from app.extensions import db
from app.models import Contract


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):
    """A contract row whose file lives in a throwaway directory."""

    def _make(player_id, uploader_id):
        stored = 'deadbeef.pdf'
        path = tmp_path / stored
        path.write_bytes(_PDF_BYTES)
        with app.app_context():
            contract = Contract(
                player_id=player_id,
                uploaded_by_id=uploader_id,
                original_filename='contract.pdf',
                stored_filename=stored,
                file_path=str(path),
            )
            db.session.add(contract)
            db.session.commit()
            return contract.id, tmp_path

    return _make


class TestSignedUpload:
    def test_an_executable_named_pdf_is_refused(
        self, app, client, as_role, make_user, contract_for
    ):
        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(NOT_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 is None
        assert not os.path.exists(directory / 'signed_deadbeef.pdf')

    def test_a_foreign_extension_is_refused(self, app, client, as_role, make_user, contract_for):
        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(b'<?php system($_GET[0]); ?>'), 'shell.php')},
            content_type='multipart/form-data',
            follow_redirects=True,
        )

        with app.app_context():
            assert db.session.get(Contract, contract_id).status != 'signed'

    def test_a_real_pdf_still_goes_through(self, app, client, as_role, make_user, contract_for):
        """Guard against over-correcting: signing a contract is the point."""
        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_at is not None
        assert os.path.exists(directory / 'signed_deadbeef.pdf')

    def test_another_player_still_cannot_sign_it(
        self, app, client, as_role, make_user, contract_for
    ):
        owner_id = make_user('player')
        admin_id = make_user('admin')
        contract_id, _directory = contract_for(owner_id, admin_id)
        as_role('player')

        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():
            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):
        from app.routes.users import ALLOWED_CONTRACT_EXTENSIONS, pdf_upload_error

        with app.test_request_context('/'):
            assert pdf_upload_error(None, ALLOWED_CONTRACT_EXTENSIONS)

    def test_it_leaves_the_stream_readable(self, app):
        """The signature check consumes bytes; the caller still has to save
        the whole file afterwards."""
        from werkzeug.datastructures import FileStorage

        from app.routes.users import ALLOWED_CONTRACT_EXTENSIONS, pdf_upload_error

        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

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