Always pin the server, let the registry own pins, and survive a revoke mid-upgrade

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.
This commit is contained in:
cproudlock
2026-09-15 15:22:25 -04:00
parent 1c2ba677c3
commit 8f10c09a76
10 changed files with 381 additions and 100 deletions

View File

@@ -168,24 +168,109 @@ def test_installer_never_removes_enrolled_credential(client, db, auth_headers, s
assert 'credential' not in line.lower(), line
def test_ca_certificate_is_not_imported_into_root(client, db, auth_headers, servercert):
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 = client.post(
PACKAGE_URL,
data={'baseurl': 'https://shopdb.test', 'collectorkey': SECRET,
'cadenceseconds': '900',
'certfile': (io.BytesIO(capem), 'ca.crt')},
content_type='multipart/form-data', headers=auth_headers)
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
# A CA cannot be the pinned leaf; the server's own leaf is pinned instead.
pins = json.loads(files['appsettings.json'])['serverpins']
assert pins == [servercert]
assert capin not 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):