shopdb-flask/tests/test_plugins/test_endpoints_server_certificate.py

"""Server TLS certificate upload and generate routes.

The new pair must be parsed, and the key proven to match the certificate,
BEFORE the existing files are backed up or replaced: a bad upload must leave
nginx's current certificate exactly where it was. cn and san must be plain
hostname/IP syntax; anything else (newlines, '=', brackets) is refused.
"""

import io
from datetime import datetime, timedelta, timezone

import pytest
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.oid import NameOID

from plugins.endpoints.api import routes as endpoint_routes

UPLOAD_URL = '/api/endpoints/settings/server-cert/upload'
GENERATE_URL = '/api/endpoints/settings/server-cert/generate'


@pytest.fixture(autouse=True)
def cert_dir(tmp_path, monkeypatch):
    directory = tmp_path / 'certs'
    directory.mkdir()
    monkeypatch.setattr(endpoint_routes, '_server_cert_dir', lambda: directory)
    monkeypatch.setattr(endpoint_routes, '_restart_nginx', lambda: None)
    return directory


def _pair(cn='old.example.test'):
    key = ec.generate_private_key(ec.SECP256R1())
    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
    now = datetime.now(timezone.utc)
    certificate = (x509.CertificateBuilder()
                   .subject_name(name).issuer_name(name)
                   .public_key(key.public_key())
                   .serial_number(x509.random_serial_number())
                   .not_valid_before(now - timedelta(days=1))
                   .not_valid_after(now + timedelta(days=30))
                   .sign(key, hashes.SHA256()))
    cert_pem = certificate.public_bytes(serialization.Encoding.PEM)
    key_pem = key.private_bytes(serialization.Encoding.PEM,
                                serialization.PrivateFormat.PKCS8,
                                serialization.NoEncryption())
    return cert_pem, key_pem


def _seed_existing(cert_dir):
    cert_pem, key_pem = _pair('existing.example.test')
    (cert_dir / 'server.crt').write_bytes(cert_pem)
    (cert_dir / 'server.key').write_bytes(key_pem)
    return cert_pem, key_pem


def _upload(client, headers, cert_bytes, key_bytes):
    return client.post(UPLOAD_URL, headers=headers,
                       content_type='multipart/form-data',
                       data={'cert': (io.BytesIO(cert_bytes), 'server.crt'),
                             'key': (io.BytesIO(key_bytes), 'server.key')})


def test_a_matching_pair_replaces_and_backs_up(client, db, auth_headers, cert_dir):
    from shopdb.api import AuditLog

    old_cert, old_key = _seed_existing(cert_dir)
    new_cert, new_key = _pair('new.example.test')

    response = _upload(client, auth_headers, new_cert, new_key)

    assert response.status_code == 200, response.get_json()
    assert (cert_dir / 'server.crt').read_bytes() == new_cert
    assert (cert_dir / 'server.key').read_bytes() == new_key
    backups = sorted(p.name for p in cert_dir.iterdir() if '.backup.' in p.name)
    assert len(backups) == 2
    assert not [p for p in cert_dir.iterdir() if p.name.endswith('.tmp')]
    assert AuditLog.query.filter_by(
        action='endpoints.server_cert.uploaded').count() == 1


def test_a_mismatched_key_touches_nothing(client, db, auth_headers, cert_dir):
    old_cert, old_key = _seed_existing(cert_dir)
    new_cert, _new_key = _pair('new.example.test')
    _other_cert, other_key = _pair('other.example.test')

    response = _upload(client, auth_headers, new_cert, other_key)

    assert response.status_code == 400
    assert 'does not match' in response.get_json()['data']['error']['message']
    assert (cert_dir / 'server.crt').read_bytes() == old_cert
    assert (cert_dir / 'server.key').read_bytes() == old_key
    assert sorted(p.name for p in cert_dir.iterdir()) == ['server.crt', 'server.key']


def test_garbage_certificate_touches_nothing(client, db, auth_headers, cert_dir):
    old_cert, _old_key = _seed_existing(cert_dir)
    _cert, key = _pair()
    garbage = b'-----BEGIN CERTIFICATE-----\nnope\n-----END CERTIFICATE-----\n'

    response = _upload(client, auth_headers, garbage, key)

    assert response.status_code == 400
    assert (cert_dir / 'server.crt').read_bytes() == old_cert
    assert len(list(cert_dir.iterdir())) == 2


@pytest.mark.parametrize('cn', [
    'host\nprompt = yes',
    'bad host',
    'a=b',
    '[alt_names]',
    'a' * 65,
])
def test_generate_refuses_unsafe_cn(client, db, auth_headers, cert_dir, cn):
    _seed_existing(cert_dir)
    response = client.post(GENERATE_URL, headers=auth_headers,
                           json={'cn': cn, 'days': 30})
    assert response.status_code == 400
    assert len(list(cert_dir.iterdir())) == 2


@pytest.mark.parametrize('san', [
    'DNS:ok.example.test,DNS:bad\nname',
    'IP:999.1.1.1',
    'URI:http://x',
    'DNS:a b',
])
def test_generate_refuses_unsafe_san(client, db, auth_headers, cert_dir, san):
    _seed_existing(cert_dir)
    response = client.post(GENERATE_URL, headers=auth_headers,
                           json={'cn': 'ok.example.test', 'san': san,
                                 'days': 30})
    assert response.status_code == 400
    assert len(list(cert_dir.iterdir())) == 2


def test_generate_installs_a_valid_pair(client, db, auth_headers, cert_dir):
    from shopdb.api import AuditLog

    _seed_existing(cert_dir)
    response = client.post(GENERATE_URL, headers=auth_headers,
                           json={'cn': 'shopdb.example.test',
                                 'san': 'DNS:shopdb.example.test,IP:10.0.0.5',
                                 'days': 30})

    assert response.status_code == 200, response.get_json()
    certificate = x509.load_pem_x509_certificate(
        (cert_dir / 'server.crt').read_bytes())
    assert certificate.subject.get_attributes_for_oid(
        NameOID.COMMON_NAME)[0].value == 'shopdb.example.test'
    assert endpoint_routes._validate_certificate_pair(
        (cert_dir / 'server.crt').read_bytes(),
        (cert_dir / 'server.key').read_bytes()) is None
    assert AuditLog.query.filter_by(
        action='endpoints.server_cert.generated').count() == 1