#!/usr/bin/env python3 """Regression tests for the certificate publication shell scripts. Why this file exists -------------------- renew-certificates.sh and sync-certificates.sh published a bundle with cat "$CERT_FILE" "$KEY_FILE" > "$COMBINED_FILE" where $COMBINED_FILE is the pem HAProxy is serving *right now*. The shell truncates the destination when it opens the redirect, before cat runs, so any failure after that point - unreadable source key, ENOSPC, container killed mid-write - left a zero-length or key-less pem behind. The exit status of cat was checked, but by then the live file was already destroyed. HAProxy loads $SSL_CERTS_DIR as a DIRECTORY (`bind :443 ssl crt /etc/haproxy/ certs`) and tries to load every file in it, so a single unloadable file fails the whole bind: HTTPS down for every customer on the host. The fix (scripts/cert-publish-lib.sh) assembles into a sibling staging dir, validates, backs up the outgoing bundle into a sibling backup dir, and renames into place. These tests pin the observable guarantees: * a successful publish replaces the live pem and archives the old one; * a FAILED publish leaves the previous, still-valid pem byte-for-byte intact; * nothing that is not a final *.pem ever appears in the certs directory; * HAProxy is not reloaded when `haproxy -c` rejects the configuration. Running ------- python3 scripts/test-cert-scripts.py # tests the repo checkout HAPROXY_MANAGER_DIR=/some/other/tree \ python3 scripts/test-cert-scripts.py # tests another tree Self-contained stdlib unittest - no pytest, no venv, no bats, and nothing is imported from the application. The scripts are driven as subprocesses with every path they touch redirected by environment variable (SSL_CERTS_DIR, LETSENCRYPT_LIVE_DIR, CERT_STAGING_DIR, CERT_BACKUP_DIR, LOG_FILE, ERROR_LOG_FILE, HAPROXY_CONFIG) and stub `certbot`, `socat` and `haproxy` binaries on PATH. The certificate material below is a real self-signed test certificate with its matching key (plus a second, unrelated key for the mismatch case), embedded as constants so the tests need no openssl to *create* material. openssl IS needed to run them: the library's cert/key pairing check is mandatory (it is the only layer that can reject a bundle of empty pem blocks), so without the binary every publish is refused by design. The image ships openssl 3.x. The acceptance bar for this file is a green run INSIDE the built image, as root, which is where these scripts actually execute - not on a workstation. Several failure modes are invisible outside the container (root ignores the directory permissions one test used to rely on) and one was actively destructive there; see _cleanup_tmp(). """ import os import re import shutil import stat import subprocess import tempfile import textwrap import time import unittest MODULE_DIR = os.path.abspath( os.environ.get('HAPROXY_MANAGER_DIR', os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')) ) SCRIPTS_DIR = os.path.join(MODULE_DIR, 'scripts') LIB = os.path.join(SCRIPTS_DIR, 'cert-publish-lib.sh') BROKEN_TOKEN = '__BROKEN__' DOMAIN = 'test.example.com' # --- test key material ------------------------------------------------------- # openssl req -x509 -newkey rsa:2048 -keyout key1 -out cert1 -days 3650 -nodes \ # -subj /CN=test.example.com TEST_CERT = """\ -----BEGIN CERTIFICATE----- MIIDFzCCAf+gAwIBAgIUeaz/lNOESOTHsB3Y97+Xja7Fy4gwDQYJKoZIhvcNAQEL BQAwGzEZMBcGA1UEAwwQdGVzdC5leGFtcGxlLmNvbTAeFw0yNjA4MDYxNTQyMTda Fw0zNjA4MDMxNTQyMTdaMBsxGTAXBgNVBAMMEHRlc3QuZXhhbXBsZS5jb20wggEi MA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDF5GL7Gjn+UnPFy5sqP2k4XHth mkWFZj+mjK6cDhbBXYt60NrwVdrrgOFydMC75VeUceFxG/5GD7wrXZP23xzbnWKm 7FxfOSmr4y+1rVEZwi8IeWEz3W6C6y5rjZsCI+pBgdna+aJSpTQZHPfDpNtQm5vl enj5BfizYixinORxm9kvXMGXV+Cw1CJkqB3mzScwWt40EtQoVxekebf8B7i4ZHyx xT6/xwF+WY8OliZkY1pdqncoTLUAYcaE/HR/ojJKmSVIq1GswZE/y3E56LIwq+wJ eGbgH46a+86z+VO2UX1jbad1kWKBCsRoOpaybZDEWAYTOqahW7kOH7umTUsRAgMB AAGjUzBRMB0GA1UdDgQWBBTIqa3BNEjjcxkhXqnRwInrLM9yijAfBgNVHSMEGDAW gBTIqa3BNEjjcxkhXqnRwInrLM9yijAPBgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3 DQEBCwUAA4IBAQB/lYzXb5PI3magMz/IXmwTsMCrVSdaUYEIKLEJggmbGxqpwO1a iYagWZ/5H3B9KDvNQA+L4FkMJ726ZkdGEH/vkwvTAuhwU2NSWcbRJ8DK5u3Q4rnJ VswPcW5njUF9mQq0NPX/PMCeOoFDEI8+RrgQZxtHhopwuKOgVA6HRBINKdEZJlrp oLLQrHDNVLMYTclNHG6kBg0lOHUV31TgkJQ8kMgtq0WQX7RseKR10QKgN5iOBomU 3+y713Ibpac5B1zw5l3LjE/59xFteFbDENr2+A5VGhVNVZC+bs+YTYTzeAsGB0e3 MQ+XJMJq3kaJmQ+QcTrRaKMtoMz2h1AIRbLd -----END CERTIFICATE----- """ # The private key that matches TEST_CERT. TEST_KEY = """\ -----BEGIN PRIVATE KEY----- MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQDF5GL7Gjn+UnPF y5sqP2k4XHthmkWFZj+mjK6cDhbBXYt60NrwVdrrgOFydMC75VeUceFxG/5GD7wr XZP23xzbnWKm7FxfOSmr4y+1rVEZwi8IeWEz3W6C6y5rjZsCI+pBgdna+aJSpTQZ HPfDpNtQm5vlenj5BfizYixinORxm9kvXMGXV+Cw1CJkqB3mzScwWt40EtQoVxek ebf8B7i4ZHyxxT6/xwF+WY8OliZkY1pdqncoTLUAYcaE/HR/ojJKmSVIq1GswZE/ y3E56LIwq+wJeGbgH46a+86z+VO2UX1jbad1kWKBCsRoOpaybZDEWAYTOqahW7kO H7umTUsRAgMBAAECggEAA9SnFFqGfR1yO4XUlfmmgyZqJoJmnl2TdZlDT4bHyrwx dSIKHO0iiNzEsHMhYHnA62EVdruunUNUdofwE28v9zHZnSbV5mt8OqUyERue5mdf gvPbjXYXu63LBx61fZH9qME3WwFqUpx7UNGiW62LJ8ktWEC5ywNCNFG+D3YfR3Iu v3PdIFXwkpJMaXO+42JSoSVoiqxlONqNiqcdQj64iYgCYdNxcgdLy+mHGrf1hAZs dom2Jp0oEM4ZdOQu3Z6uyEyPsECiz1PdQjzagaEfPWtKCQk0kY8DDCn5X2xQif9w 3xeaISqho7bQgSo4IEgWHTP6V0v7brTP7VcK+gyMyQKBgQD15FrE4XAL89NBQ6iX m2l6quZv5tIUtreuYxsOpFIMU22zfsRZOvA7sJR3ufOJ+b7tFHuJ8DAJUpK08Vvg A930/LW9wUsY42d54XKIO/8DTsIrmjppoGdshs3axJQkqfN0zQkhqfbwaXRJ9X2k Fvax+5jftaIaw0hQdWLnfTmGmQKBgQDOBue89D8THYl/LvDGg6WoVyK3ljBMK2s3 4BljeZCJdWAjtido13Mltubc6YVHScmVoIZKTmx+fCjdQ3y1t92vZZSBeLsXhfFr N+IOGZu3ZmJu64x3OukSYQ7x6agi5yP3+7k0siZgxOMXJRQDYZUcHxIHDrSGLeLZ sj7LnbvLOQKBgE8murE1gEPYsOAJT3O96y45ZQQQYP+Z8XaJIGSOMHsXP/DPlZTD jCEqrh/8E5EOe48FUN8OGehmVCM6rkBl/kSmNDpoxiu0x9JL5/pClcwSxh4S/0qQ /7nHiuwo6ycCLgQjHBViCMNKrsw/4bm4SqDwRD1+0jebNOPxZWzuul3BAoGAZij0 ZjSyxhbCZEdxau5CiYvTkjct8cch3k4IKNRRwGdsaajcN9eFqHDeXzKIPQYwqDo1 /MiQcdO9K6JYR39JtLxo/B5Sn2JyiJjoRdea6EEjlB7GwyR6B/wKvhf/oHb+1euD NccU0q0ucf6XwulzV8NsXAWFrHc6YnpJOwwW37kCgYAkEyjUc73jImiTyf/IXVOD UHlRXZPvwtZUuPGe4RI0Gds97tKnvXnvFsPIRCOGfVzZ8z79DGiQ8TR2a0hgZec1 Mo3J2dCjlv4Q6ACjHkCA1cmi13OHUPnpaeesrOk+SpEVJfR2k4qRWH4z3oxbPO/q TDFnjWHSgjkDMrxVHZ8wPg== -----END PRIVATE KEY----- """ # A perfectly valid RSA key that has nothing to do with TEST_CERT. UNRELATED_KEY = """\ -----BEGIN PRIVATE KEY----- MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDE7mosrsdBhdn1 ZIErYmMuPU69ws51JdyqwtRZlV2uLLby0dOfJpFKZPuBPEqTqDimd9N4FSL0CGjG zXG23RYd9AbdSoorFORrUxeiPzFnbz4v38srGHckOS9ozKAbPLACUEjWMX1NwAZ+ wGlSz3cWYTWVztFCUIxpvLVR87PSpTnpdCXIj0EABOc6WoLBLE+v2knZOa63LKg6 GryInD43CnWFBKpH0gdgWqh+ie3NFMumLR8M3lZq2Mk0EFgVWrxPnJobYvOmavNp NBR3ZugR4X57c1YFyLkYXQIdhxuYV1ZAY6NgmAIsaiCKFqdksF5AJSV7FZNBM43r 33ae/R5hAgMBAAECggEAIgNYsL9+OEWtU8ooWi0r2q5plXJaVNb1gkPUx+U5sS3V amKNxbj0UrBW1Scr7U1afXwIOP8TkqkKKb4NpCsS2RkO/3USoKbC3fuTwyjdeFM5 Hy0sytR2rXm4A8aF57ZnYvrpXZ9eGEnwhT9n4Y7mL2YaSnXWZDkDy3Z1rcIk/p5P jUo4UQyzD/9Gab1stcBbGv+66B3mlVdRVJor5+tGn6zmr4TjvqupAexuXd+Q6+1L OG4e2bAUmuOVOa+w8Xo4vwkiXSDLVEsl1z1x4Bcrv61bIg0rbZAeqcZ4EUPyrEyR RpaaOLAiwdzjOj5pQvHAF2q/++IlY6xPhotTr11tvQKBgQD37C2UQRGRo622+2kx qgsdA6jPM9vCE9S4aS4qaOj8XSMMCu6bQW5lU48DaONvXxDmT6x9IrgR3QN1Iz/t 17kUCCghuviEP1RjnahBRQqZR38KdDzKhaWt9jgdjCun9MzUXIaQRTFlFGA1z1pz apZ9a8/ehYPSk4pv9h06/B3IjQKBgQDLWPBuhj203QOmwVF1v1k4yyuJ7UBqtRaY I9jMW93sB2hPs6Se10UroQHlF95IHeRXvNKH/UXuAILIJMN8oTI4uU7t60shJHvI o14RzEZoUQjES7BBVhePglndJmYIKoKKAX5lSFe9Lk0ei9B+1Dle5Q+9ZeP09cp4 vVcGvOgqJQKBgQCtxe1srO8TlhZ821uwY+/GNnpsQX0XW68OUyr4rvAfc2jNWBxG 1mX6v8bOLQa9WXUO+Wl9jIhYfQGfaUW2AC7Jy63VdqgaigksiaUVmr8DEQoK2c6C ZYrrlFlg3I78+qlXcEMhfF5S6yVEkkJkA6HX52mcHxl2z9OJBokWfwChQQKBgAzH xzy7FS/D4FHfvpXu89Wc91yQ28aZIRVo01xsvbLy+DxiJwuQrhlC4lKawG656jsV dAn2AiomQBICNYMkwnpMM0jCzBMGLv16PxRRSW+PAEUOGMLSfWKYp7s9iZYjzdaM p3wIIvOR8GjmErGV9xEexnF58OzZceNKyyhyQQk9AoGAbvjJvYLOhQpAbbqErNNv LK1P+TngKpukRHXjiUpPVEGNhN6krBBJCBWzY7ucrIy6jz8UBy7SbITBy7qIKhxZ PVP7WVATMWEeW1AfdBfCYDI4jFKAD8SLECby45nRBuBllYdQnW1gBzLcCulCwB+w FV0RvuQPDYkqsx8ibqpSv7c= -----END PRIVATE KEY----- """ # A SECOND, unrelated but internally consistent pair. Needed by the # concurrency test: two bundles that are each perfectly valid but whose keys # differ, so a validator that reads the certificate from one and the key from # the other reports a mismatch. Two bundles sharing key material - which is # what PREVIOUS_BUNDLE and NEW_BUNDLE are - cannot detect that at all. # openssl req -x509 -newkey rsa:2048 -nodes -days 3650 \ # -subj /CN=second.example.com TEST_CERT_2 = """\ -----BEGIN CERTIFICATE----- MIIDGzCCAgOgAwIBAgIUKQAAvrVkden7gIg2zYMLb6dO1c8wDQYJKoZIhvcNAQEL BQAwHTEbMBkGA1UEAwwSc2Vjb25kLmV4YW1wbGUuY29tMB4XDTI2MDgwNjE2NTcz NVoXDTM2MDgwMzE2NTczNVowHTEbMBkGA1UEAwwSc2Vjb25kLmV4YW1wbGUuY29t MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAv2pfTOvN2zrZJ6d+biiD VOczMqanuHPBCy2CnTBif+7VWf8AaywRzQ3ShfhmalVRNEFMn0MSO+GocmH71Ve1 nl89oGiJPmvX1lpncbgM692ddhP9ez4xUeNj+QAWp9VBZhInNuM4Pawv5BPngtpj 2MGXf3ZlBSli8Ng7jBo1fTMT3bh8GcE1rIPRvmUuQwFIt2eGnLR8jQd+xGelhAjG nnXtlc+ebo4r2OjljNgvtdUknBZdpiZXmjdFzyClYTeMuEen2uwMpJNc0wLbRjcU khVF3nw4jUnkOhWH3JYGAoWslJyEqZSANwt/eOHwXgyVuxg31bCl297iskW0IRZL wQIDAQABo1MwUTAdBgNVHQ4EFgQU7HH8GacJzc2j9s2UJCPwykgrIckwHwYDVR0j BBgwFoAU7HH8GacJzc2j9s2UJCPwykgrIckwDwYDVR0TAQH/BAUwAwEB/zANBgkq hkiG9w0BAQsFAAOCAQEAYE5jHX1dK091jVsFSZDdiw9AU5rrk8XpF1yuPmDisRnE dJ4QQq3dzWXRnp0bzZnq7fdfiEz1m39zVixov7WFp24QhenD2n5K7/wew7RpXTnA pAGBEdsGvBJ+3MgkRYklXCM9f9f4z21xXRNZ+BwBcM25D+gR4b+PRMQR6BhZx5R+ y2jQsoM68cFjRApFWgmji4pBjg/eOaZMBCfVTjP+npVyqG7UtV5EyYXwPgPa/rm2 FoP+eftzP6dszBEonkIVyyvkdscI4Wkr8hw3S0R/TP8l9lTnvN1HC3o7Es5VY53R swNAXWBlgm0N7A96ISLtQjgvOfeMRTCSjxW9pm0wJA== -----END CERTIFICATE----- """ TEST_KEY_2 = """\ -----BEGIN PRIVATE KEY----- MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQC/al9M683bOtkn p35uKINU5zMypqe4c8ELLYKdMGJ/7tVZ/wBrLBHNDdKF+GZqVVE0QUyfQxI74ahy YfvVV7WeXz2gaIk+a9fWWmdxuAzr3Z12E/17PjFR42P5ABan1UFmEic24zg9rC/k E+eC2mPYwZd/dmUFKWLw2DuMGjV9MxPduHwZwTWsg9G+ZS5DAUi3Z4actHyNB37E Z6WECMaede2Vz55ujivY6OWM2C+11SScFl2mJleaN0XPIKVhN4y4R6fa7Aykk1zT AttGNxSSFUXefDiNSeQ6FYfclgYChayUnISplIA3C3944fBeDJW7GDfVsKXb3uKy RbQhFkvBAgMBAAECggEABQLqy6eedRPt31sLOMFEDkzbZdFHOMpZeuqThsNmjo7z t9diokgeD4ZQXimbEQqsZYDAtmFGdiEp86I9JQBk/4TUiqFhHrOADBe1jQAptjfs iyIb51gIPnjZ1RBA5Al8zohy28T9h1+Z7+/OeCvcgLyVAixf/U9pU8D1El9cu9zv 4q4WJPB5Tgkq+YwcmeuT8LzsKoSDmPQjFVY9v+gz6hoVUVyP6gswnlFnKjNcmfZU 0CKP90sCAc5mKZv7RyGG920LDU4u2ggnQoK05GhXK8R3amJmoAF3i+xGeTRvNEKe wDC7NinTG2WDVo6y/FCvFKs0+qqlKww1u39fq166rQKBgQDtjPjvkDkSYFcWFznn sfsxN5R1cLpxdutLZhfvtpHIj5NkQbmOk0r3LbDAwU+UNVQ6F8jWQVHUsni9jAY6 3RNRF4ZabC51aqg/Ssj7d5mE4kj7y6Ch/nnSXIogRxLG4bWo2rtG1M2Dt3Ef08R4 6gjcp9ZmELyVx7H4yXFJGIaDdQKBgQDOSCHFK/ggZzzy4iC7DOq1lrbR+jyk95rh F5EGzyAkgJ4uYc9TCPkUDxXjWL17r61/obfbW4znaV9fHEKpz3R54osDXln3oLea BBWkJI3ANe8iNrGDE4FN9to4DdUMFsWX/WEiRyBPIqDy9DTyadblMLTNZjNPuLmg ZJOB3tRZnQKBgQCjeglya9Eq2Uv1MuSxk2VnmHU9YOed4BXLHKZKXFz1JgFr1GNL QAguFK536FDIkO62z9lxwR/8fRnkb7F13uBFRSg7oAlU2qKQc/nePI9UyJkrVxXj hYn2f6K61c6ROZFXc7e/5gDMrXhXS9gA0iZpG8PLF6eAeB39NTwV7p/bZQKBgQCV v5eEY58FJu0ABVhtcbsRiA+/70EHIRi2Pz1xC/vxg81RLoArb2AiR7FEEa+8kpQJ C4VFIPjxJXWuvf1G+Os9cFAqadw1/95JWJ29QywEVSL8W2gSF57O0l0oRCJdXEql Q7O4BppV2HWu6clmEZ+HUgxu77pgLWHUJi9PIExXoQKBgEq6rYBXmsSQermnPTOn YFx7c2ns97hsjYIbs497+gPW4/xQWwsN76t60SWqjXV4DHJCpi5Tnjo2fk5/OzK+ HfshC9CKFDk8T0KGQWwaPwqP/OYbqOA88IlJ7xbPdSuJkjefHCtVVEtao6+HNgzm lniLtDMpU0MLPgB98ClQ4HDA -----END PRIVATE KEY----- """ # The bundle already on disk when a run starts. Same key material, plus a # trailing marker so "the live file was replaced" and "the old file was # archived" can be told apart byte-for-byte. PREVIOUS_BUNDLE = TEST_CERT + TEST_KEY + '# previous bundle\n' NEW_BUNDLE = TEST_CERT + TEST_KEY OTHER_BUNDLE = TEST_CERT_2 + TEST_KEY_2 # --- stub binaries ----------------------------------------------------------- STUB_CERTBOT = """\ #!/bin/sh # Test stub for certbot: pretend there was nothing to renew. echo "No renewals were attempted." exit 0 """ STUB_SOCAT = """\ #!/bin/sh # Test stub for socat: record that a reload was attempted, and what was sent. { printf 'socat %s <<' "$*"; cat; printf '>>\\n'; } >> "$SOCAT_LOG" exit 0 """ STUB_HAPROXY = """\ #!/bin/sh # Test stub for the haproxy binary: `haproxy -c -f FILE` rejects any config # containing %(token)s, which is how the tests inject an invalid config. cfg="" while [ $# -gt 0 ]; do case "$1" in -f) cfg="$2"; shift ;; esac shift done if [ -n "$cfg" ] && grep -q '%(token)s' "$cfg" 2>/dev/null; then echo "[ALERT] parsing [$cfg:1] : unknown keyword '%(token)s'" >&2 exit 1 fi exit 0 """ % {'token': BROKEN_TOKEN} GOOD_HAPROXY_CFG = textwrap.dedent("""\ global daemon defaults mode http frontend fe bind 0.0.0.0:443 ssl crt /etc/haproxy/certs """) def write(path, content, mode=None): with open(path, 'w') as fh: fh.write(content) if mode is not None: os.chmod(path, mode) return path def read(path): with open(path) as fh: return fh.read() class CertScriptFixture(unittest.TestCase): """An isolated fake /etc/haproxy + /etc/letsencrypt plus stub binaries.""" def setUp(self): # Without the library every "the library rejects X" assertion in this # file can be satisfied by bash exiting 127, so make its absence a # failure of every test rather than a silent pass of several. self.assertTrue(os.path.isfile(LIB), f'{LIB} is missing - nothing below tests anything') self.tmp = tempfile.mkdtemp(prefix='haproxy-cert-test-') self.addCleanup(self._cleanup_tmp) self.bindir = os.path.join(self.tmp, 'bin') os.makedirs(self.bindir) write(os.path.join(self.bindir, 'certbot'), STUB_CERTBOT, 0o755) write(os.path.join(self.bindir, 'socat'), STUB_SOCAT, 0o755) write(os.path.join(self.bindir, 'haproxy'), STUB_HAPROXY, 0o755) self.socat_log = os.path.join(self.tmp, 'socat-invocations.log') # Mirrors the real layout: certs dir, with staging/backups as SIBLINGS. self.haproxy_dir = os.path.join(self.tmp, 'etc', 'haproxy') self.certs_dir = os.path.join(self.haproxy_dir, 'certs') self.staging_dir = os.path.join(self.haproxy_dir, 'cert-staging') self.backup_dir = os.path.join(self.haproxy_dir, 'cert-backups') os.makedirs(self.certs_dir) self.le_live = os.path.join(self.tmp, 'etc', 'letsencrypt', 'live') self.domain_dir = os.path.join(self.le_live, DOMAIN) os.makedirs(self.domain_dir) self.src_cert = write(os.path.join(self.domain_dir, 'fullchain.pem'), TEST_CERT) self.src_key = write(os.path.join(self.domain_dir, 'privkey.pem'), TEST_KEY) self.live_pem = os.path.join(self.certs_dir, DOMAIN + '.pem') self.haproxy_cfg = write(os.path.join(self.haproxy_dir, 'haproxy.cfg'), GOOD_HAPROXY_CFG) self.log_file = os.path.join(self.tmp, 'haproxy-manager.log') self.error_log = os.path.join(self.tmp, 'haproxy-manager-errors.log') def _cleanup_tmp(self): # A test may have chmod 000'd a fixture file, which would stop rmtree. # # SYMLINKS ARE SKIPPED, and that is not a nicety. os.chmod() FOLLOWS # symlinks, and the openssl-availability tests below build a stripped # PATH directory out of symlinks to real system binaries (/usr/bin/cat, # /usr/bin/chmod, ...). Walking those with chmod 0600 as root - which is # how this container runs - stripped the exec bit from a dozen core # binaries of the machine running the tests, chmod itself included, so # it could not even be undone from inside the container: every later # test failed with "/usr/bin/grep: Permission denied" and certificate # publishing stayed dead until the container was recreated. It never # showed up on a workstation because an unprivileged chmod of a # root-owned file fails EPERM and was swallowed by `except OSError`. # This file ships in the image (COPY scripts /haproxy/scripts), so # running it in place is a thing an operator will do. for root, dirs, files in os.walk(self.tmp): for name in files: path = os.path.join(root, name) if os.path.islink(path): continue try: os.chmod(path, 0o600) except OSError: pass shutil.rmtree(self.tmp, ignore_errors=True) # -- helpers --------------------------------------------------------- def env(self, **overrides): env = dict(os.environ) env.update({ 'PATH': self.bindir + os.pathsep + os.environ['PATH'], 'SSL_CERTS_DIR': self.certs_dir, 'LETSENCRYPT_LIVE_DIR': self.le_live, 'CERT_STAGING_DIR': self.staging_dir, 'CERT_BACKUP_DIR': self.backup_dir, 'LOG_FILE': self.log_file, 'ERROR_LOG_FILE': self.error_log, 'HAPROXY_CONFIG': self.haproxy_cfg, 'SOCAT_LOG': self.socat_log, }) env.update(overrides) return env def run_script(self, name, **env_overrides): path = os.path.join(SCRIPTS_DIR, name) self.assertTrue(os.path.exists(path), f'{path} does not exist') return subprocess.run(['bash', path], env=self.env(**env_overrides), capture_output=True, text=True, timeout=120) def bundle_is_valid(self, path): """Ask the shipped library whether HAProxy could use this bundle.""" return subprocess.run( ['bash', '-c', '. "$1"; cert_bundle_valid "$2"', '_', LIB, path], env=self.env(), capture_output=True, text=True).returncode == 0 def reload_attempted(self): return os.path.exists(self.socat_log) and os.path.getsize(self.socat_log) > 0 def logs(self): text = '' for path in (self.log_file, self.error_log): if os.path.exists(path): text += read(path) return text def seed_previous_bundle(self, content=PREVIOUS_BUNDLE): return write(self.live_pem, content) def assert_certs_dir_is_clean(self): """HAProxy loads this directory wholesale: only final *.pem may be here.""" entries = sorted(os.listdir(self.certs_dir)) strays = [e for e in entries if not e.endswith('.pem')] self.assertEqual(strays, [], f'non-.pem files left in the certs directory HAProxy ' f'loads wholesale: {strays} (dir: {entries})') def assert_no_staging_leftovers(self): if os.path.isdir(self.staging_dir): self.assertEqual(sorted(os.listdir(self.staging_dir)), [], 'staging file was not cleaned up') def assert_rejected(self, result, because): """The library refused, FOR THE STATED REASON. `assertNotEqual(rc, 0)` on its own proves nothing about this library. Delete cert-publish-lib.sh and `. "$1"` fails, cert_bundle_valid is never defined, bash exits 127 - and a bare rc!=0 assertion passes. Four tests in TestCertPublishLibrary were doing exactly that; they were pinning "some bash pipeline failed", not "the bundle was rejected". """ output = result.stdout + result.stderr self.assertNotIn('command not found', output, 'the shell could not find the function under test - ' 'this asserts nothing about the library') self.assertNotEqual(127, result.returncode, f'exit 127 means "no such command", not "rejected": {output}') self.assertNotEqual(0, result.returncode, f'expected a rejection, got success: {output}') self.assertIn(because, output, f'rejected, but not for the expected reason ' f'({because!r} not in output): {output}') class CertScriptBehaviour: """Behaviour shared by renew-certificates.sh and sync-certificates.sh. A mixin rather than a TestCase so the cases are collected once per concrete script, not a third time for the base class. """ SCRIPT = None def run_it(self, **env_overrides): return self.run_script(self.SCRIPT, **env_overrides) # -- happy path ------------------------------------------------------ def test_publish_replaces_live_pem_and_archives_the_previous_one(self): self.seed_previous_bundle() result = self.run_it() self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual(read(self.live_pem), NEW_BUNDLE, 'live pem is not the newly assembled cert+key') backup = os.path.join(self.backup_dir, DOMAIN + '.pem') self.assertTrue(os.path.exists(backup), f'previous bundle was not archived to {backup}') self.assertEqual(read(backup), PREVIOUS_BUNDLE, 'the archived bundle is not the one that was replaced') self.assertIn('1 updated, 0 failed', self.logs()) self.assertTrue(self.reload_attempted(), 'HAProxy was never reloaded after a successful update') self.assert_certs_dir_is_clean() self.assert_no_staging_leftovers() def test_first_publish_works_with_no_previous_bundle(self): result = self.run_it() self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual(read(self.live_pem), NEW_BUNDLE) self.assert_certs_dir_is_clean() # -- THE HEADLINE ---------------------------------------------------- def test_unreadable_source_key_leaves_the_previous_bundle_intact(self): """The bug this whole change exists for. Pre-fix: `cat cert key > live.pem` truncated live.pem before cat ran, so an unreadable key left a key-less (or empty) pem in the directory HAProxy loads wholesale -> the :443 bind fails -> every site on the host loses HTTPS. Checking cat's exit status did not undo that. """ self.seed_previous_bundle() before = read(self.live_pem) self.assertTrue(self.bundle_is_valid(self.live_pem), 'fixture precondition: the seeded bundle must be valid') how = self.break_source_key() result = self.run_it() self.assertEqual(read(self.live_pem), before, f'the live pem was damaged by a failed publish ({how}); ' f'HAProxy would fail to load the certs directory') self.assertTrue(self.bundle_is_valid(self.live_pem), 'the pem left on disk is no longer a usable bundle') logs = self.logs() self.assertIn(f'Failed to combine certificate for {DOMAIN}', logs) self.assertIn('0 updated, 1 failed', logs) self.assertFalse(self.reload_attempted(), 'HAProxy was reloaded even though nothing was updated') self.assert_certs_dir_is_clean() self.assert_no_staging_leftovers() # This used to assert returncode == 0 with the comment "a per-domain # failure should not change the exit code", codifying the script's # `exit 0`. That is wrong and it is the dangerous kind of wrong: a host # where EVERY domain fails to publish looked, to cron and to # host-renew-certificates.sh (which branches on this exit code), # exactly like a clean run. Nothing would notice until the certificates # expired. Continuing past a failed domain so the others still get # published is right; reporting success afterwards is not. self.assertNotEqual(result.returncode, 0, 'a domain that failed to publish must be reported ' 'in the exit code, not just in the log') def break_source_key(self): """Make reading the source key fail, however this environment allows. chmod 000 is the faithful reproduction (file present, `-f` true, cat fails), but it is a no-op for root, so as root we truncate the key instead: pre-fix that is even nastier, because `cat` then *succeeds* and silently publishes a key-less pem. """ if os.geteuid() == 0: write(self.src_key, '') return 'zero-length source key (running as root)' os.chmod(self.src_key, 0o000) return 'unreadable source key (chmod 000)' # -- other ways to end up with an unusable bundle -------------------- def test_cert_only_bundle_is_rejected(self): self.seed_previous_bundle() before = read(self.live_pem) write(self.src_key, TEST_CERT) # no private key block at all self.run_it() self.assertEqual(read(self.live_pem), before, 'a key-less bundle was published over the live pem') self.assertIn('0 updated, 1 failed', self.logs()) self.assertFalse(self.reload_attempted()) self.assert_certs_dir_is_clean() self.assert_no_staging_leftovers() def test_truncated_certificate_block_is_rejected(self): self.seed_previous_bundle() before = read(self.live_pem) write(self.src_cert, TEST_CERT.split('\n')[0] + '\nMIIDFzCCAf+gAwIBA\n') self.run_it() self.assertEqual(read(self.live_pem), before, 'a truncated certificate was published over the live pem') self.assertIn('0 updated, 1 failed', self.logs()) self.assert_certs_dir_is_clean() def test_mismatched_key_is_rejected(self): if shutil.which('openssl') is None: self.skipTest('openssl CLI not available: without it every publish ' 'is refused, so this test could not tell a pairing ' 'rejection from a missing-checker rejection') self.seed_previous_bundle() before = read(self.live_pem) write(self.src_key, UNRELATED_KEY) self.run_it() self.assertEqual(read(self.live_pem), before, 'a bundle whose key does not match the cert was published') self.assertIn('does not match the certificate', self.logs()) self.assertIn('0 updated, 1 failed', self.logs()) self.assert_certs_dir_is_clean() # -- the certs directory is HAProxy's, not ours ---------------------- def test_no_stray_files_in_certs_dir_after_success_or_failure(self): self.seed_previous_bundle() self.run_it() self.assert_certs_dir_is_clean() self.assertEqual(sorted(os.listdir(self.certs_dir)), [DOMAIN + '.pem']) self.break_source_key() self.run_it() self.assert_certs_dir_is_clean() self.assertEqual(sorted(os.listdir(self.certs_dir)), [DOMAIN + '.pem']) self.assert_no_staging_leftovers() def test_staging_and_backup_dirs_are_outside_the_certs_dir(self): """Belt and braces: even with the defaults, nothing lands under certs/.""" self.seed_previous_bundle() # Drop the explicit overrides so the derived defaults are exercised. env = {'CERT_STAGING_DIR': '', 'CERT_BACKUP_DIR': ''} self.run_it(**env) self.assert_certs_dir_is_clean() self.assertEqual(sorted(os.listdir(self.certs_dir)), [DOMAIN + '.pem']) self.assertTrue( os.path.exists(os.path.join(self.haproxy_dir, 'cert-staging')), 'default staging dir is not the documented sibling of the certs dir') self.assertTrue( os.path.exists(os.path.join(self.haproxy_dir, 'cert-backups', DOMAIN + '.pem')), 'default backup dir is not the documented sibling of the certs dir') # -- reload gating --------------------------------------------------- def test_reload_is_not_attempted_when_haproxy_config_is_invalid(self): write(self.haproxy_cfg, GOOD_HAPROXY_CFG + BROKEN_TOKEN + '\n') result = self.run_it() self.assertFalse(self.reload_attempted(), 'HAProxy was reloaded with a configuration that ' '`haproxy -c` rejects') self.assertNotEqual(result.returncode, 0, 'refusing to reload must be a loud, non-zero exit') self.assertIn('does not validate', self.logs()) def test_reload_happens_when_the_config_validates(self): self.run_it() self.assertTrue(self.reload_attempted()) self.assertIn('reload', read(self.socat_log)) def test_reload_is_not_attempted_when_nothing_was_updated(self): shutil.rmtree(self.domain_dir) result = self.run_it() self.assertEqual(result.returncode, 0) self.assertFalse(self.reload_attempted()) class TestRenewCertificates(CertScriptBehaviour, CertScriptFixture): SCRIPT = 'renew-certificates.sh' class TestSyncCertificates(CertScriptBehaviour, CertScriptFixture): SCRIPT = 'sync-certificates.sh' class TestCertPublishLibrary(CertScriptFixture): """Unit-level checks on cert-publish-lib.sh itself.""" def call(self, snippet, *args, **env_overrides): return subprocess.run( ['bash', '-c', '. "$1"; shift; ' + snippet, '_', LIB, *args], env=self.env(**env_overrides), capture_output=True, text=True) def test_valid_bundle_accepted(self): path = write(os.path.join(self.tmp, 'ok.pem'), NEW_BUNDLE) self.assertEqual(self.call('cert_bundle_valid "$1"', path).returncode, 0) def test_empty_and_missing_bundles_rejected(self): empty = write(os.path.join(self.tmp, 'empty.pem'), '') self.assert_rejected(self.call('cert_bundle_valid "$1"', empty), 'is empty') missing = os.path.join(self.tmp, 'nope.pem') self.assert_rejected(self.call('cert_bundle_valid "$1"', missing), 'does not exist') def test_key_without_end_marker_rejected(self): truncated = write(os.path.join(self.tmp, 'cut.pem'), TEST_CERT + '-----BEGIN PRIVATE KEY-----\nMIIEvAIB\n') self.assert_rejected(self.call('cert_bundle_valid "$1"', truncated), 'unterminated private key block') def test_empty_pem_blocks_are_rejected(self): """Why the pairing check is mandatory rather than best-effort. Every structural check in the library passes on this file: a complete CERTIFICATE block and a complete PRIVATE KEY block, both with nothing between BEGIN and END. Only openssl can tell it is not a certificate. """ hollow = write(os.path.join(self.tmp, 'hollow.pem'), '-----BEGIN CERTIFICATE-----\n' '-----END CERTIFICATE-----\n' '-----BEGIN PRIVATE KEY-----\n' '-----END PRIVATE KEY-----\n') self.assert_rejected(self.call('cert_bundle_valid "$1"', hollow), 'openssl could not read the certificate') def test_a_broken_live_pem_does_not_overwrite_a_good_backup(self): """Mirrors create_backup(require_valid=True) in haproxy_manager.py. If the pem currently on disk is already garbage, archiving it would replace a restorable backup with an unusable one. """ os.makedirs(self.backup_dir) good_backup = write(os.path.join(self.backup_dir, DOMAIN + '.pem'), PREVIOUS_BUNDLE) write(self.live_pem, 'garbage, not a pem at all\n') result = self.call('cert_publish "$1" "$2" "$3"', self.src_cert, self.src_key, self.live_pem) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual(read(self.live_pem), NEW_BUNDLE) self.assertEqual(read(good_backup), PREVIOUS_BUNDLE, 'a good backup was overwritten with an unusable pem') def test_publish_fails_loudly_when_the_rename_cannot_happen(self): """No silent fallback to writing straight into the certs dir. The failure is injected with a stub `mv` that refuses, rather than by chmod 0500 on the certs dir: the container these scripts run in is root, root ignores directory permissions, so the chmod version skipped itself exactly where it matters and only ever ran on a workstation. """ self.seed_previous_bundle() before = read(self.live_pem) write(os.path.join(self.bindir, 'mv'), "#!/bin/sh\n" "echo \"mv: cannot move '$2': Permission denied\" >&2\n" "exit 1\n", 0o755) self.addCleanup(os.unlink, os.path.join(self.bindir, 'mv')) result = self.call('cert_publish "$1" "$2" "$3"', self.src_cert, self.src_key, self.live_pem) self.assert_rejected(result, 'NOT falling back to a direct write') self.assertEqual(read(self.live_pem), before, 'the live pem was damaged by a failed rename') self.assert_no_staging_leftovers() def test_published_pem_keeps_the_mode_of_the_file_it_replaces(self): """A write-safety fix must not silently re-permission private keys. Both directions matter: mktemp stages at 0600, so without the explicit chmod every publish would tighten a 0644 bundle; and the mode must not be copied from a symlink (see the next test). """ for mode in (0o644, 0o640, 0o600): with self.subTest(oct(mode)): write(self.live_pem, PREVIOUS_BUNDLE, mode) result = self.call('cert_publish "$1" "$2" "$3"', self.src_cert, self.src_key, self.live_pem) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual(read(self.live_pem), NEW_BUNDLE) self.assertEqual( stat.S_IMODE(os.stat(self.live_pem).st_mode), mode, 'publishing changed who can read the private key') def test_symlinked_live_pem_does_not_become_world_writable(self): """`stat -c %a` on a symlink reports 0777 - the LINK's mode, not a permission. Copying that onto the staged bundle put a world-writable private key in the directory HAProxy serves from. -L is what makes the preserved mode the mode of the file an operator actually chose. """ target = write(os.path.join(self.tmp, 'real-bundle.pem'), PREVIOUS_BUNDLE, 0o640) os.symlink(target, self.live_pem) result = self.call('cert_publish "$1" "$2" "$3"', self.src_cert, self.src_key, self.live_pem) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) mode = stat.S_IMODE(os.stat(self.live_pem).st_mode) self.assertEqual( 0, mode & 0o022, f'published bundle is group/world writable ({oct(mode)}) - the ' f'symlink mode was copied onto a real private key') self.assertEqual(0o640, mode) def test_publish_refuses_when_staging_is_on_another_filesystem(self): """The header used to claim a cross-device mv "fails loudly and leaves the live pem alone". GNU mv does no such thing: across filesystems it copies, so it truncates and writes the DESTINATION first and only then discovers it cannot finish (ENOSPC being the realistic case) - the very truncation this library exists to prevent. Both directories are env-overridable, so the device numbers have to be checked up front. """ staging = os.path.join(self.a_dir_on_another_filesystem(), 'cert-staging') self.seed_previous_bundle() before = read(self.live_pem) result = self.call('cert_publish "$1" "$2" "$3"', self.src_cert, self.src_key, self.live_pem, CERT_STAGING_DIR=staging) self.assert_rejected(result, 'different filesystems') self.assertEqual(read(self.live_pem), before, 'the live pem was disturbed by a refused publish') self.assert_certs_dir_is_clean() def test_stale_python_side_staging_temps_are_reaped(self): """The staging dir has two writers. write_config_atomically() on the Python side stages as `..tmp` (tempfile.mkstemp(prefix=name + '.', suffix='.tmp')). The reaper matched only mktemp's `*.??????` shape, so every temp leaked by a SIGKILL on the Python side stayed there forever. """ os.makedirs(self.staging_dir, exist_ok=True) stale_py = write(os.path.join(self.staging_dir, DOMAIN + '.pem.ab12cd34.tmp'), 'stale\n') stale_sh = write(os.path.join(self.staging_dir, DOMAIN + '.pem.AbCdEf'), 'stale\n') fresh = write(os.path.join(self.staging_dir, 'recent.pem.zz99yy.tmp'), 'fresh\n') old = time.time() - 3 * 24 * 3600 for path in (stale_py, stale_sh): os.utime(path, (old, old)) result = self.call('cert_publish "$1" "$2" "$3"', self.src_cert, self.src_key, self.live_pem) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertFalse(os.path.exists(stale_py), 'a stale Python-side staging temp was never reaped') self.assertFalse(os.path.exists(stale_sh), 'a stale shell-side staging temp was never reaped') self.assertTrue(os.path.exists(fresh), 'the reaper deleted a temp a concurrent publisher may ' 'still be writing') def test_validation_of_a_live_pem_being_republished_is_not_spurious(self): """cert_bundle_valid must judge ONE snapshot of the file. cert_publish() calls cert_bundle_valid() on the LIVE pem (step (e), to decide whether it is worth backing up) while another publisher may be renaming a new bundle over it. The function used to open the file six times, so `openssl x509` could read the outgoing bundle and `openssl pkey` the incoming one - and report "private key does not match the certificate" about two files that were each perfectly fine. That ERROR goes into the log monitor-errors.sh watches, which makes it a page. The two bundles alternated below are each internally valid but carry DIFFERENT key material. That matters: NEW_BUNDLE and PREVIOUS_BUNDLE share a cert and a key, so alternating those two could never produce a mismatch no matter how badly the reads were interleaved - the test would model a world in which the bug cannot happen and pass forever. """ write(self.live_pem, NEW_BUNDLE) a = write(os.path.join(self.tmp, 'churn-a.pem'), NEW_BUNDLE) b = write(os.path.join(self.tmp, 'churn-b.pem'), OTHER_BUNDLE) # A publisher renaming over the live pem as fast as it can. Staged # outside the certs dir, then renamed, exactly like cert_publish(). churn = subprocess.Popen( ['bash', '-c', 'end=$((SECONDS+8)); s="$4"; while [ $SECONDS -lt $end ]; do ' ' cp "$1" "$s"; mv -f "$s" "$2"; ' ' cp "$3" "$s"; mv -f "$s" "$2"; ' 'done', '_', a, self.live_pem, b, os.path.join(self.tmp, 'churn-staged.pem')], env=self.env(), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def _stop(): churn.kill() churn.wait() self.addCleanup(_stop) result = subprocess.run( ['bash', '-c', '. "$1"; for i in $(seq 1 200); do cert_bundle_valid "$2" || exit 1; done', '_', LIB, self.live_pem], env=self.env(), capture_output=True, text=True, timeout=120) _stop() output = result.stdout + result.stderr self.assertNotIn('does not match the certificate', output, 'two valid bundles were reported as a mismatched pair ' 'because the checks read different files') self.assertEqual(0, result.returncode, f'a concurrent republish made validation fail: {output}') def a_dir_on_another_filesystem(self): certs_dev = os.stat(self.certs_dir).st_dev for candidate in ('/dev/shm', '/run', '/var/tmp', '/tmp', '/'): try: if (os.path.isdir(candidate) and os.access(candidate, os.W_OK) and os.stat(candidate).st_dev != certs_dev): path = tempfile.mkdtemp(prefix='cert-xdev-', dir=candidate) self.addCleanup(shutil.rmtree, path, True) return path except OSError: continue self.skipTest('no writable directory on a second filesystem available') def test_haproxy_config_ok_follows_the_validator(self): self.assertEqual(self.call('haproxy_config_ok').returncode, 0) write(self.haproxy_cfg, GOOD_HAPROXY_CFG + BROKEN_TOKEN + '\n') self.assertNotEqual(self.call('haproxy_config_ok').returncode, 0) def _openssl_free_path(self, name): """A PATH directory with the library's tools but no openssl. Symlinks, so this directory must never be walked with a chmod that follows them - see _cleanup_tmp(). """ fake_path = os.path.join(self.tmp, name) os.makedirs(fake_path) for tool in ('cat', 'grep', 'mktemp', 'mv', 'cp', 'rm', 'mkdir', 'basename', 'dirname', 'find', 'date', 'chmod', 'stat'): real = shutil.which(tool) if real: os.symlink(real, os.path.join(fake_path, tool)) return fake_path def test_missing_openssl_is_a_hard_failure(self): """The pairing check is MANDATORY: no openssl, no publication. This used to assert the opposite - that a missing openssl warns and publishes anyway - justified by "the image does not necessarily install the openssl CLI". The image does: openssl 3.x arrives with ca-certificates, which certbot needs, and generate_self_signed_cert() already runs `openssl req` with check=True at first-run setup. So the fail-open never actually fired, and structural checks alone accept a bundle of empty pem blocks (see test_empty_pem_blocks_are_rejected). """ fake_path = self._openssl_free_path('no-openssl-bin') path = write(os.path.join(self.tmp, 'ok.pem'), NEW_BUNDLE) # bash by absolute path: the stripped PATH cannot resolve it. result = subprocess.run( [shutil.which('bash'), '-c', '. "$1"; cert_bundle_valid "$2"', '_', LIB, path], env=self.env(PATH=fake_path), capture_output=True, text=True) self.assert_rejected(result, 'openssl binary not found') self.assertIn('REFUSING', result.stdout + result.stderr, 'a broken image must be reported as a broken image') def test_missing_openssl_stops_a_publish_rather_than_weakening_it(self): """cert_publish must inherit the refusal, and not touch the live pem.""" fake_path = self._openssl_free_path('no-openssl-bin2') self.seed_previous_bundle() before = read(self.live_pem) result = subprocess.run( [shutil.which('bash'), '-c', '. "$1"; cert_publish "$2" "$3" "$4"', '_', LIB, self.src_cert, self.src_key, self.live_pem], env=self.env(PATH=fake_path), capture_output=True, text=True) self.assert_rejected(result, 'openssl binary not found') self.assertEqual(read(self.live_pem), before, 'the live pem was disturbed by a refused publish') self.assert_certs_dir_is_clean() self.assert_no_staging_leftovers() class TestScriptsAreSane(unittest.TestCase): """Cheap static guards against the failure mode coming back.""" SHELL_FILES = ('renew-certificates.sh', 'sync-certificates.sh', 'cert-publish-lib.sh') def test_shell_files_parse(self): for name in self.SHELL_FILES: path = os.path.join(SCRIPTS_DIR, name) result = subprocess.run(['bash', '-n', path], capture_output=True, text=True) self.assertEqual(result.returncode, 0, f'{name}: {result.stderr}') def test_no_script_redirects_into_the_live_pem(self): pattern = re.compile(r'>\s*"?\$\{?COMBINED_FILE') for name in ('renew-certificates.sh', 'sync-certificates.sh'): body = read(os.path.join(SCRIPTS_DIR, name)) self.assertIsNone(pattern.search(body), f'{name} still redirects output straight into the ' f'live pem HAProxy is serving') if __name__ == '__main__': print(f'testing scripts from: {SCRIPTS_DIR}') unittest.main(verbosity=2)