shopdb-flask/tests/test_plugins/test_endpoints_siteagent_build.py

"""Site agent package build: server pins, optional shared key, Root import gate.

The package is the only place a PC gets its first trust anchor and (optionally)
the shared fleet key, so these tests read the actual ZIP the route returns
rather than the helper in isolation.
"""

import datetime
import hashlib
import io
import json
import zipfile

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 agentstore, siteagent
from shopdb.core.models.setting import Setting

PACKAGE_URL = '/api/endpoints/siteagent/package'
STATUS_URL = '/api/endpoints/siteagent'
SECRET = 'sharedkeyvalue-0123456789abcdef'


def _make_cert(isca):
    key = ec.generate_private_key(ec.SECP256R1())
    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,
                                         'ca.test' if isca else 'server.test')])
    now = datetime.datetime.now(datetime.timezone.utc)
    builder = (x509.CertificateBuilder()
               .subject_name(name).issuer_name(name)
               .public_key(key.public_key())
               .serial_number(x509.random_serial_number())
               .not_valid_before(now - datetime.timedelta(days=1))
               .not_valid_after(now + datetime.timedelta(days=30)))
    if isca:
        builder = builder.add_extension(
            x509.BasicConstraints(ca=True, path_length=None), critical=True)
    certificate = builder.sign(key, hashes.SHA256())
    pem = certificate.public_bytes(serialization.Encoding.PEM)
    spki = certificate.public_key().public_bytes(
        serialization.Encoding.DER,
        serialization.PublicFormat.SubjectPublicKeyInfo)
    return pem, hashlib.sha256(spki).hexdigest()


@pytest.fixture(autouse=True)
def isolate(tmp_path, monkeypatch):
    """Empty agent store and no auto-detected server certificate by default."""
    store = tmp_path / 'siteagent'
    store.mkdir()
    monkeypatch.setattr(agentstore, '_siteagent_dir', lambda: str(store))
    monkeypatch.setattr(siteagent, '_CERT_CANDIDATES', ())
    return tmp_path


@pytest.fixture
def servercert(tmp_path, monkeypatch):
    """A self-signed leaf staged as this server's own certificate."""
    pem, pin = _make_cert(isca=False)
    path = tmp_path / 'server.crt'
    path.write_bytes(pem)
    monkeypatch.setattr(siteagent, '_CERT_CANDIDATES', (str(path),))
    return pin


def _build(client, headers, **fields):
    body = {'baseurl': 'https://shopdb.test', 'cadenceseconds': 900}
    body.update(fields)
    return client.post(PACKAGE_URL, json=body, headers=headers)


def _files(response):
    assert response.status_code == 200, response.get_data(as_text=True)[:400]
    archive = zipfile.ZipFile(io.BytesIO(response.data))
    return {name: archive.read(name) for name in archive.namelist()}


def test_installer_writes_current_server_pin(client, db, auth_headers, servercert):
    files = _files(_build(client, auth_headers, collectorkey=SECRET,
                          includesharedkey=True, bundlecert=True))
    script = files['install.ps1'].decode('ascii')

    assert '[string[]] $ServerPins = @("{0}")'.format(servercert) in script
    assert 'Set-HklmMultiStringBothViews -SubKey "SOFTWARE\\GE\\ShopDB" -Name "ServerPins"' in script
    assert '[Microsoft.Win32.RegistryValueKind]::MultiString' in script
    assert '-ceq $wanted' in script
    settings = json.loads(files['appsettings.json'])
    assert settings['serverpins'] == [servercert]


def test_pin_comes_from_server_cert_even_when_not_bundled(
        client, db, auth_headers, servercert):
    files = _files(_build(client, auth_headers, collectorkey=SECRET))
    assert 'server.crt' not in files
    assert json.loads(files['appsettings.json'])['serverpins'] == [servercert]


def test_next_pin_is_shipped_beside_current(client, db, auth_headers, servercert):
    nextpin = 'ab' * 32
    files = _files(_build(client, auth_headers, collectorkey=SECRET,
                          nextserverpin=nextpin))
    script = files['install.ps1'].decode('ascii')

    assert '@("{0}", "{1}")'.format(servercert, nextpin) in script
    assert json.loads(files['appsettings.json'])['serverpins'] == [servercert, nextpin]


@pytest.mark.parametrize('badpin', ['AB' * 32, 'ab' * 31, 'zz' * 32])
def test_next_pin_must_be_64_lowercase_hex(client, db, auth_headers, servercert, badpin):
    response = _build(client, auth_headers, collectorkey=SECRET, nextserverpin=badpin)
    assert response.status_code == 400


def test_next_pin_without_a_current_pin_is_refused(client, db, auth_headers):
    response = _build(client, auth_headers, collectorkey=SECRET,
                      nextserverpin='ab' * 32)
    assert response.status_code == 400


def test_shared_key_on_keeps_key_provisioning(client, db, auth_headers, servercert):
    files = _files(_build(client, auth_headers, collectorkey=SECRET,
                          includesharedkey='true'))
    script = files['install.ps1'].decode('ascii')

    assert '[string] $CollectorKey = "{0}"'.format(SECRET) in script
    assert '-Name "CollectorKey" -Value $CollectorKey' in script
    assert 'Settings > Endpoints > Enrollment' not in script


def test_shared_key_off_leaves_no_key_anywhere(client, db, auth_headers, servercert):
    files = _files(_build(client, auth_headers, collectorkey=SECRET,
                          includesharedkey=False))

    for name, content in files.items():
        assert b'CollectorKey' not in content, name
        assert SECRET.encode('ascii') not in content, name
    script = files['install.ps1'].decode('ascii')
    assert 'Settings > Endpoints > Enrollment' in script
    assert '$sharedKeyMode = "omitted"' in script
    assert json.loads(files['appsettings.json'])['apikey'] == ''
    assert b'Settings > Endpoints > Enrollment' in files['README.txt']


def test_shared_key_default_follows_site_setting(client, db, auth_headers, servercert):
    # Setting on (default): key required.
    assert _build(client, auth_headers).status_code == 400

    Setting.set('plugin.endpoints.sharedkeyallowed', False, valuetype='boolean')
    status = client.get(STATUS_URL, headers=auth_headers).get_json()['data']
    assert status['sharedkeyallowed'] is False
    assert status['serverpin'] == servercert

    files = _files(_build(client, auth_headers, collectorkey=SECRET))
    assert b'CollectorKey' not in files['install.ps1']
    assert SECRET.encode('ascii') not in files['install.ps1']


def test_installer_never_removes_enrolled_credential(client, db, auth_headers, servercert):
    script = _files(_build(client, auth_headers, collectorkey=SECRET))[
        'install.ps1'].decode('ascii')
    assert 'existing enrolled credential found; kept as is.' in script
    for line in script.splitlines():
        if 'Remove-Item' in line:
            assert 'credential' not in line.lower(), line


def _upload(client, headers, pem, filename, **fields):
    data = {'baseurl': 'https://shopdb.test', 'collectorkey': SECRET,
            'cadenceseconds': '900', 'certfile': (io.BytesIO(pem), filename)}
    data.update(fields)
    return client.post(PACKAGE_URL, data=data,
                       content_type='multipart/form-data', headers=headers)


def test_ca_certificate_is_pinned_but_not_imported_into_root(
        client, db, auth_headers):
    capem, capin = _make_cert(isca=True)
    response = _upload(client, auth_headers, capem, 'ca.crt')
    files = _files(response)
    script = files['install.ps1'].decode('ascii')

    assert 'server.crt' in files
    assert '$trustCertInRoot = $false' in script
    assert 'is a CA certificate; not importing it' in script
    assert 'Install-ShopDBServerCertificate.ps1' in script
    assert 'older than 1.20' in script
    # The SPKI pin ignores basicConstraints: the uploaded leaf is pinned.
    assert '[string[]] $ServerPins = @("{0}")'.format(capin) in script
    assert json.loads(files['appsettings.json'])['serverpins'] == [capin]
    assert response.headers.get(siteagent.CA_WARNING_HEADER) == \
        siteagent.CA_CERT_BUILD_WARNING


def test_uploaded_ca_certificate_ships_both_pins_when_server_cert_differs(
        client, db, auth_headers, servercert):
    capem, capin = _make_cert(isca=True)
    nextpin = 'ef' * 32
    response = _upload(client, auth_headers, capem, 'ca.crt',
                       nextserverpin=nextpin)
    files = _files(response)
    script = files['install.ps1'].decode('ascii')

    assert '[string[]] $ServerPins = @("{0}", "{1}", "{2}")'.format(
        capin, servercert, nextpin) in script
    assert json.loads(files['appsettings.json'])['serverpins'] == [
        capin, servercert, nextpin]
    assert '$trustCertInRoot = $false' in script
    warning = response.headers.get(siteagent.CA_WARNING_HEADER)
    assert siteagent.CA_CERT_BUILD_WARNING in warning
    assert 'both pins were included' in warning


def test_uploaded_ca_certificate_matching_server_cert_ships_one_pin(
        client, db, auth_headers, tmp_path, monkeypatch):
    capem, capin = _make_cert(isca=True)
    path = tmp_path / 'server.crt'
    path.write_bytes(capem)
    monkeypatch.setattr(siteagent, '_CERT_CANDIDATES', (str(path),))
    response = _upload(client, auth_headers, capem, 'server.crt')
    files = _files(response)

    assert json.loads(files['appsettings.json'])['serverpins'] == [capin]
    assert 'both pins' not in response.headers.get(siteagent.CA_WARNING_HEADER)


@pytest.fixture
def caservercert(tmp_path, monkeypatch):
    """A CA-flagged self-signed cert staged as this server's own certificate,
    as a plain openssl req -x509 produces."""
    pem, pin = _make_cert(isca=True)
    path = tmp_path / 'server.crt'
    path.write_bytes(pem)
    monkeypatch.setattr(siteagent, '_CERT_CANDIDATES', (str(path),))
    return pin


def test_ca_flagged_server_cert_bundled_is_still_pinned(
        client, db, auth_headers, caservercert):
    nextpin = 'cd' * 32
    response = _build(client, auth_headers, collectorkey=SECRET,
                      bundlecert=True, nextserverpin=nextpin)
    files = _files(response)
    script = files['install.ps1'].decode('ascii')

    assert 'server.crt' in files
    assert '[string[]] $ServerPins = @("{0}", "{1}")'.format(caservercert, nextpin) in script
    assert json.loads(files['appsettings.json'])['serverpins'] == [caservercert, nextpin]
    assert '$trustCertInRoot = $false' in script
    warning = response.headers.get(siteagent.CA_WARNING_HEADER)
    assert warning and 'Install-ShopDBServerCertificate.ps1' in warning
    warning.encode('ascii')


def test_ca_flagged_server_cert_not_bundled_is_still_pinned(
        client, db, auth_headers, caservercert):
    response = _build(client, auth_headers, collectorkey=SECRET)
    files = _files(response)

    assert 'server.crt' not in files
    assert json.loads(files['appsettings.json'])['serverpins'] == [caservercert]
    assert response.headers.get(siteagent.CA_WARNING_HEADER)
    status = client.get(STATUS_URL, headers=auth_headers).get_json()['data']
    assert status['serverpin'] == caservercert


def test_leaf_certificate_build_has_no_warning(client, db, auth_headers, servercert):
    response = _build(client, auth_headers, collectorkey=SECRET, bundlecert=True)
    _files(response)
    assert siteagent.CA_WARNING_HEADER not in response.headers


def test_leaf_certificate_is_imported_into_root(client, db, auth_headers):
    leafpem, leafpin = _make_cert(isca=False)
    response = client.post(
        PACKAGE_URL,
        data={'baseurl': 'https://shopdb.test', 'collectorkey': SECRET,
              'cadenceseconds': '900',
              'certfile': (io.BytesIO(leafpem), 'server.crt')},
        content_type='multipart/form-data', headers=auth_headers)
    files = _files(response)
    script = files['install.ps1'].decode('ascii')

    assert '$trustCertInRoot = $true' in script
    assert 'Import-Certificate -FilePath $certPath -CertStoreLocation Cert:\\LocalMachine\\Root' in script
    assert json.loads(files['appsettings.json'])['serverpins'] == [leafpin]


def test_installer_restricts_the_registry_key_to_system_and_admins(
        client, db, auth_headers, servercert):
    """The shared key stays in HKLM until every consumer is on a per-PC
    credential, so what matters meanwhile is who can read it. The key inherits a
    DACL that grants BUILTIN\\Users read, which is the exposure the per-PC
    migration exists to close - the installer must not leave it that way."""
    files = _files(_build(client, auth_headers, collectorkey=SECRET,
                          includesharedkey='true'))
    script = files['install.ps1'].decode('ascii')

    assert 'function Protect-HklmShopdbKey' in script
    assert 'Protect-HklmShopdbKey -SubKey "SOFTWARE\\GE\\ShopDB"' in script
    # SYSTEM and Administrators by SID, so a localised Windows still matches.
    assert '"S-1-5-18", "S-1-5-32-544"' in script
    # Inheritance broken WITHOUT copying the inherited entries: copying them is
    # what silently preserves the Users read.
    assert 'SetAccessRuleProtection($true, $false)' in script
    assert 'RemoveAccessRule($rule)' in script


def test_the_key_is_written_before_it_is_locked(client, db, auth_headers, servercert):
    """SetAccessControl replaces the DACL rather than merging, so a lock that ran
    first would be undone by nothing - but a write that ran after a lock the
    installer no longer holds rights for would fail. Order is the guarantee."""
    script = _files(_build(client, auth_headers, collectorkey=SECRET,
                           includesharedkey='true'))['install.ps1'].decode('ascii')

    write = script.index('-Name "CollectorKey" -Value $CollectorKey')
    lock = script.index('Protect-HklmShopdbKey -SubKey')
    assert write < lock


def test_the_registry_lock_failing_does_not_fail_the_install(
        client, db, auth_headers, servercert):
    """A bay that cannot take the ACL still needs its agent. It must say plainly
    that the key is readable there, rather than dying or staying quiet."""
    script = _files(_build(client, auth_headers, collectorkey=SECRET,
                           includesharedkey='true'))['install.ps1'].decode('ascii')

    lock = script.index('Protect-HklmShopdbKey -SubKey')
    tail = script[lock:lock + 600]
    assert 'catch' in tail
    assert 'readable by any logged-on user' in tail


def test_the_keyless_package_still_locks_the_key(client, db, auth_headers, servercert):
    """ServerPins live in the same key and the registry is authoritative for
    pinning, so a standard user must not be able to rewrite them either."""
    files = _files(_build(client, auth_headers, collectorkey=SECRET,
                          includesharedkey=False))
    script = files['install.ps1'].decode('ascii')

    assert b'CollectorKey' not in files['install.ps1']
    assert 'Protect-HklmShopdbKey -SubKey "SOFTWARE\\GE\\ShopDB"' in script


def test_install_script_helper_without_pins_or_key():
    script = siteagent._install_script('https://shopdb.test', '', 900,
                                       serverpins=[], includesharedkey=False)
    assert '[string[]] $ServerPins = @()' in script
    assert 'CollectorKey' not in script
    assert 'ServerPins left as is' in script