8f10c09a76
Found in the Windows VM re-test. - A CA-flagged server certificate (common with a plain openssl self-signed cert) produced an installer package with no pin at all, so agent 1.20 refused the server. The package now always carries the SPKI pin of the certificate PCs receive; the CA flag only skips the trusted-root import. When an uploaded CA-flagged certificate differs from the server's own, both pins are included. The build response carries X-ShopDB-Build-Warning and the installer logs what older agents need. - Pins written into appsettings.json by the installer were merged with the registry, so removing a pin from HKLM never took effect. When the ServerPins value exists in the registry it is now authoritative, even when empty; appsettings pins apply only when it is absent. Set-ShopDBServerPins.ps1 -Clear deletes the value; new -None writes an empty value (no pins). - A script or package download sent with a credential revoked moments earlier got 401 CREDENTIAL_INVALID and failed the command. The agent now retries that download once with the shared key and records the fallback in the result.
298 lines
12 KiB
Python
298 lines
12 KiB
Python
"""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_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
|