From 2e45c3f2e51e550885fdb5a1934b28048baadaf0 Mon Sep 17 00:00:00 2001 From: Joao Gilberto Magalhaes Date: Sat, 14 Feb 2026 16:59:22 -0500 Subject: [PATCH] Add unit and integration tests for Certbot and HAProxy configuration - Added integration tests to validate HAProxy configuration generation for Certbot's HTTP-01 challenges. - Verified handling of ACLs, SSL redirects, multi-domain setups, and backend behaviors. - Introduced unit tests for the Certbot class to ensure proper ACME server handling, certificate status checks, and configuration validation. - Improved test coverage for edge cases, including EAB credentials, manual hooks, and custom ports. - Enhanced certbot-related environment variable parsing and error scenarios. --- tests/test_certbot.py | 765 +++++++++++++++++++++++++++ tests/test_certbot_haproxy_config.py | 414 +++++++++++++++ 2 files changed, 1179 insertions(+) create mode 100644 tests/test_certbot.py create mode 100644 tests/test_certbot_haproxy_config.py diff --git a/tests/test_certbot.py b/tests/test_certbot.py new file mode 100644 index 0000000..6db70f4 --- /dev/null +++ b/tests/test_certbot.py @@ -0,0 +1,765 @@ +""" +Unit tests for Certbot/ACME functionality + +Tests the Certbot class without requiring internet access or third-party providers. +Verifies command generation, certificate status checking, and configuration handling. +""" + +import logging +import os +import sys +import tempfile +import time +from datetime import datetime, timedelta +from unittest.mock import MagicMock, Mock, mock_open, patch + +from OpenSSL import crypto + +# Add src to path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from functions import Certbot, ContainerEnv, Functions + + +class TestCertbotStaticMethods: + """Test Certbot static helper methods""" + + def test_set_acme_server_empty(self): + """Test ACME server with empty string""" + assert Certbot.set_acme_server("") == "" + assert Certbot.set_acme_server(None) == "" + assert Certbot.set_acme_server(False) == "" + + def test_set_acme_server_staging(self): + """Test ACME server with staging flag""" + assert Certbot.set_acme_server("staging") == "--staging" + assert Certbot.set_acme_server("STAGING") == "--staging" + assert Certbot.set_acme_server("Staging") == "--staging" + + def test_set_acme_server_custom_url(self): + """Test ACME server with custom URL""" + url = "https://acme-v02.api.letsencrypt.org/directory" + assert Certbot.set_acme_server(url) == f"--server {url}" + + url2 = "https://acme.ssl.com/sslcom-dv-rsa" + assert Certbot.set_acme_server(url2) == f"--server {url2}" + + # HTTP URLs should also work + url3 = "http://localhost:14000/dir" + assert Certbot.set_acme_server(url3) == f"--server {url3}" + + def test_set_acme_server_invalid(self): + """Test ACME server with invalid values""" + assert Certbot.set_acme_server("production") == "" + assert Certbot.set_acme_server("invalid") == "" + assert Certbot.set_acme_server("test") == "" + + def test_set_eab_kid_empty(self): + """Test EAB KID with empty string""" + assert Certbot.set_eab_kid("") == "" + + def test_set_eab_kid_with_value(self): + """Test EAB KID with valid value""" + kid = "test-kid-12345" + assert Certbot.set_eab_kid(kid) == f'--eab-kid "{kid}"' + + def test_set_eab_hmac_key_empty(self): + """Test EAB HMAC key with empty string""" + assert Certbot.set_eab_hmac_key("") == "" + + def test_set_eab_hmac_key_with_value(self): + """Test EAB HMAC key with valid value""" + hmac = "test-hmac-key-abcdef" + assert Certbot.set_eab_hmac_key(hmac) == f'--eab-hmac-key "{hmac}"' + + +class TestCertbotInitialization: + """Test Certbot class initialization""" + + def test_certbot_init_basic(self): + """Test Certbot initialization with basic configuration""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_SERVER': 'staging', + }, clear=False): + certbot = Certbot("/tmp/certs") + + assert certbot.certs == "/tmp/certs" + assert certbot.email == "test@example.com" + assert certbot.acme_server == "--staging" + assert certbot.eab_kid == "" + assert certbot.eab_hmac_key == "" + assert certbot.freeze_issue == {} + assert certbot.retry_count == 60 # default + assert certbot.certbot_preferred_challenges == "http" # default + assert certbot.certbot_manual_auth_hook == False # default + + def test_certbot_init_with_eab(self): + """Test Certbot initialization with EAB credentials""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_SERVER': 'https://acme.ssl.com/sslcom-dv-rsa', + 'EASYHAPROXY_CERTBOT_EAB_KID': 'my-eab-kid', + 'EASYHAPROXY_CERTBOT_EAB_HMAC_KEY': 'my-hmac-key', + }, clear=False): + certbot = Certbot("/tmp/certs") + + assert certbot.email == "test@example.com" + assert certbot.acme_server == "--server https://acme.ssl.com/sslcom-dv-rsa" + assert certbot.eab_kid == '--eab-kid "my-eab-kid"' + assert certbot.eab_hmac_key == '--eab-hmac-key "my-hmac-key"' + + def test_certbot_init_with_custom_retry_count(self): + """Test Certbot initialization with custom retry count""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_RETRY_COUNT': '120', + }, clear=False): + certbot = Certbot("/tmp/certs") + + assert certbot.retry_count == 120 + + def test_certbot_init_with_dns_challenge(self): + """Test Certbot initialization with DNS challenge""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_PREFERRED_CHALLENGES': 'dns', + }, clear=False): + certbot = Certbot("/tmp/certs") + + assert certbot.certbot_preferred_challenges == "dns" + + def test_certbot_init_with_manual_auth_hook(self): + """Test Certbot initialization with manual auth hook""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_MANUAL_AUTH_HOOK': '/path/to/auth-hook.sh', + }, clear=False): + certbot = Certbot("/tmp/certs") + + assert certbot.certbot_manual_auth_hook == "/path/to/auth-hook.sh" + + +class TestCertbotCertificateStatus: + """Test certificate status checking""" + + def create_test_certificate(self, days_valid=30): + """Helper to create a test certificate valid for specified days""" + # Create key pair + key = crypto.PKey() + key.generate_key(crypto.TYPE_RSA, 2048) + + # Create certificate + cert = crypto.X509() + cert.get_subject().CN = "test.example.com" + cert.set_serial_number(1000) + cert.gmtime_adj_notBefore(0) + cert.gmtime_adj_notAfter(days_valid * 24 * 60 * 60) + cert.set_issuer(cert.get_subject()) + cert.set_pubkey(key) + cert.sign(key, 'sha256') + + # Combine cert and key + cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert) + key_pem = crypto.dump_privatekey(crypto.FILETYPE_PEM, key) + + return cert_pem.decode() + key_pem.decode() + + def test_get_certificate_status_not_found(self): + """Test certificate status when file doesn't exist""" + with patch.dict(os.environ, {'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com'}, clear=False): + certbot = Certbot("/tmp/nonexistent") + status = certbot.get_certificate_status("example.com") + assert status == "not_found" + + def test_get_certificate_status_ok(self): + """Test certificate status when valid and not expiring soon""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.pem', delete=False) as f: + cert_content = self.create_test_certificate(days_valid=90) + f.write(cert_content) + cert_file = f.name + + try: + with patch.dict(os.environ, {'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com'}, clear=False): + certbot = Certbot(os.path.dirname(cert_file)) + # Mock the filename pattern + with patch.object(certbot, 'certs', os.path.dirname(cert_file)): + status = certbot.get_certificate_status(os.path.basename(cert_file).replace('.pem', '')) + assert status == "ok" + finally: + os.unlink(cert_file) + + def test_get_certificate_status_expiring(self): + """Test certificate status when expiring within 15 days""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.pem', delete=False) as f: + cert_content = self.create_test_certificate(days_valid=10) + f.write(cert_content) + cert_file = f.name + + try: + with patch.dict(os.environ, {'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com'}, clear=False): + certbot = Certbot(os.path.dirname(cert_file)) + with patch.object(certbot, 'certs', os.path.dirname(cert_file)): + status = certbot.get_certificate_status(os.path.basename(cert_file).replace('.pem', '')) + assert status == "expiring" + finally: + os.unlink(cert_file) + + def test_get_certificate_status_expired(self): + """Test certificate status when already expired""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.pem', delete=False) as f: + cert_content = self.create_test_certificate(days_valid=-1) + f.write(cert_content) + cert_file = f.name + + try: + with patch.dict(os.environ, {'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com'}, clear=False): + certbot = Certbot(os.path.dirname(cert_file)) + with patch.object(certbot, 'certs', os.path.dirname(cert_file)): + status = certbot.get_certificate_status(os.path.basename(cert_file).replace('.pem', '')) + assert status == "expired" + finally: + os.unlink(cert_file) + + def test_get_certificate_status_error(self): + """Test certificate status with invalid/corrupted certificate""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.pem', delete=False) as f: + f.write("Invalid certificate content\n") + cert_file = f.name + + try: + with patch.dict(os.environ, {'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com'}, clear=False): + certbot = Certbot(os.path.dirname(cert_file)) + with patch.object(certbot, 'certs', os.path.dirname(cert_file)): + status = certbot.get_certificate_status(os.path.basename(cert_file).replace('.pem', '')) + assert status == "error" + finally: + os.unlink(cert_file) + + +class TestCertbotMergeCertificate: + """Test certificate merging functionality""" + + def test_merge_certificate(self): + """Test merging certificate and key into single file""" + cert = "-----BEGIN CERTIFICATE-----\nCERT_DATA\n-----END CERTIFICATE-----\n" + key = "-----BEGIN PRIVATE KEY-----\nKEY_DATA\n-----END PRIVATE KEY-----\n" + + with tempfile.NamedTemporaryFile(mode='w', delete=False) as f: + filename = f.name + + try: + Certbot.merge_certificate(cert, key, filename) + + with open(filename, 'r') as f: + content = f.read() + + assert content == cert + key + assert "BEGIN CERTIFICATE" in content + assert "BEGIN PRIVATE KEY" in content + finally: + os.unlink(filename) + + +class TestCertbotCheckCertificates: + """Test check_certificates method and command generation""" + + def test_check_certificates_no_email(self): + """Test that no certificates are requested without email""" + with patch.dict(os.environ, {'EASYHAPROXY_CERTBOT_EMAIL': ''}, clear=False): + certbot = Certbot("/tmp/certs") + result = certbot.check_certificates(["example.com"]) + assert result is False + + def test_check_certificates_no_hosts(self): + """Test that no certificates are requested without hosts""" + with patch.dict(os.environ, {'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com'}, clear=False): + certbot = Certbot("/tmp/certs") + result = certbot.check_certificates([]) + assert result is False + + @patch('functions.Functions.run_bash') + def test_check_certificates_request_new(self, mock_run_bash): + """Test requesting new certificates (not_found status)""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_SERVER': 'staging', + }, clear=False): + certbot = Certbot("/tmp/certs") + + # Mock get_certificate_status to return not_found + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + result = certbot.check_certificates(['example.com', 'test.com']) + + assert result is True + assert mock_run_bash.called + call_args = mock_run_bash.call_args[0] + command = call_args[1] + + # Verify command structure + assert '/usr/bin/certbot certonly' in command + assert '--staging' in command + assert '--preferred-challenges http' in command + assert '--agree-tos' in command + assert '--issuance-timeout 90' in command + assert '--no-eff-email' in command + assert '--non-interactive' in command + assert '--max-log-backups=0' in command + assert '-d example.com' in command + assert '-d test.com' in command + assert '--email test@example.com' in command + assert '--http-01-port 2080' in command + assert '--standalone' in command + + @patch('functions.Functions.run_bash') + def test_check_certificates_with_eab(self, mock_run_bash): + """Test certificate request with EAB credentials""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_SERVER': 'https://acme.ssl.com/sslcom-dv-rsa', + 'EASYHAPROXY_CERTBOT_EAB_KID': 'my-kid', + 'EASYHAPROXY_CERTBOT_EAB_HMAC_KEY': 'my-hmac', + }, clear=False): + certbot = Certbot("/tmp/certs") + + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + result = certbot.check_certificates(['example.com']) + + assert result is True + call_args = mock_run_bash.call_args[0] + command = call_args[1] + + assert '--eab-kid "my-kid"' in command + assert '--eab-hmac-key "my-hmac"' in command + assert '--server https://acme.ssl.com/sslcom-dv-rsa' in command + + @patch('functions.Functions.run_bash') + def test_check_certificates_with_dns_challenge(self, mock_run_bash): + """Test certificate request with DNS challenge""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_PREFERRED_CHALLENGES': 'dns', + }, clear=False): + certbot = Certbot("/tmp/certs") + + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + result = certbot.check_certificates(['example.com']) + + assert result is True + call_args = mock_run_bash.call_args[0] + command = call_args[1] + + assert '--preferred-challenges dns' in command + # DNS challenge should NOT include --http-01-port or --standalone + assert '--http-01-port' not in command + assert '--standalone' not in command + + @patch('functions.Functions.run_bash') + def test_check_certificates_with_manual_auth_hook(self, mock_run_bash): + """Test certificate request with manual auth hook""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_MANUAL_AUTH_HOOK': '/path/to/hook.sh', + }, clear=False): + certbot = Certbot("/tmp/certs") + + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + result = certbot.check_certificates(['example.com']) + + assert result is True + call_args = mock_run_bash.call_args[0] + command = call_args[1] + + assert "--manual --manual-auth-hook '/path/to/hook.sh'" in command + + @patch('functions.Functions.run_bash') + def test_check_certificates_renew(self, mock_run_bash): + """Test renewing expiring certificates""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + certbot = Certbot("/tmp/certs") + + with patch.object(certbot, 'get_certificate_status', return_value='expiring'): + with patch.object(certbot, 'find_live_certificates'): + result = certbot.check_certificates(['example.com']) + + assert result is True + assert mock_run_bash.called + + # Should call certbot renew + call_args = mock_run_bash.call_args[0] + command = call_args[1] + assert '/usr/bin/certbot renew' in command + + @patch('functions.Functions.run_bash') + def test_check_certificates_mixed_statuses(self, mock_run_bash): + """Test with mixed certificate statuses (new, renew, ok)""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + certbot = Certbot("/tmp/certs") + + # Mock different statuses for different hosts + def mock_status(host): + statuses = { + 'new.com': 'not_found', + 'renew.com': 'expiring', + 'ok.com': 'ok', + 'error.com': 'error', + } + return statuses.get(host, 'ok') + + with patch.object(certbot, 'get_certificate_status', side_effect=mock_status): + with patch.object(certbot, 'find_live_certificates'): + result = certbot.check_certificates(['new.com', 'renew.com', 'ok.com', 'error.com']) + + assert result is True + # Should be called twice: once for certonly (new.com), once for renew (renew.com) + assert mock_run_bash.call_count == 2 + + @patch('functions.Functions.run_bash') + def test_check_certificates_freeze_mechanism(self, mock_run_bash): + """Test freeze mechanism when certificate issuance fails""" + mock_run_bash.return_value = (1, []) # Return error code + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_RETRY_COUNT': '5', + }, clear=False): + certbot = Certbot("/tmp/certs") + + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + with patch.object(certbot, 'find_missing_certificates') as mock_find_missing: + result = certbot.check_certificates(['example.com']) + + assert result is True # Still returns True (reload needed) + assert mock_find_missing.called + + @patch('functions.Functions.run_bash') + def test_check_certificates_debug_mode(self, mock_run_bash): + """Test that verbose flag is added in debug mode""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'CERTBOT_LOG_LEVEL': 'DEBUG', + }, clear=False): + certbot = Certbot("/tmp/certs") + + # Set logger to DEBUG + from functions import logger_certbot + with patch.object(logger_certbot, 'level', logging.DEBUG): + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + result = certbot.check_certificates(['example.com']) + + assert result is True + call_args = mock_run_bash.call_args[0] + command = call_args[1] + + # Should include -v for verbose output + assert ' -v' in command + + +class TestCertbotFindLiveCertificates: + """Test finding and merging live certificates""" + + def test_find_live_certificates_no_directory(self): + """Test when /etc/letsencrypt/live doesn't exist""" + with patch.dict(os.environ, {'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com'}, clear=False): + certbot = Certbot("/tmp/certs") + + with patch('os.path.exists', return_value=False): + certbot.find_live_certificates() + # Should not crash + + def test_find_live_certificates_with_certs(self): + """Test finding and merging certificates from live directory""" + with tempfile.TemporaryDirectory() as tmpdir: + # Create mock directory structure + live_dir = os.path.join(tmpdir, "live") + os.makedirs(live_dir) + + # Create example.com certificate + example_dir = os.path.join(live_dir, "example.com") + os.makedirs(example_dir) + + cert_content = "-----BEGIN CERTIFICATE-----\nCERT\n-----END CERTIFICATE-----\n" + key_content = "-----BEGIN PRIVATE KEY-----\nKEY\n-----END PRIVATE KEY-----\n" + + with open(os.path.join(example_dir, "cert.pem"), 'w') as f: + f.write(cert_content) + with open(os.path.join(example_dir, "privkey.pem"), 'w') as f: + f.write(key_content) + + output_dir = os.path.join(tmpdir, "output") + os.makedirs(output_dir) + + with patch.dict(os.environ, {'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com'}, clear=False): + certbot = Certbot(output_dir) + + with patch('os.path.exists', return_value=True): + with patch('os.listdir', return_value=['example.com']): + with patch('os.path.isdir', return_value=True): + with patch.object(Functions, 'load', side_effect=[cert_content, key_content]): + certbot.find_live_certificates() + + # Verify merged certificate was created + merged_file = os.path.join(output_dir, "example.com.pem") + if os.path.exists(merged_file): + with open(merged_file, 'r') as f: + content = f.read() + assert content == cert_content + key_content + + +class TestCertbotFindMissingCertificates: + """Test freeze mechanism for failed certificates""" + + def test_find_missing_certificates_sets_freeze(self): + """Test that missing certificates are frozen for retry""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_RETRY_COUNT': '10', + }, clear=False): + certbot = Certbot("/tmp/certs") + + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + certbot.find_missing_certificates(['-d example.com', '-d test.com']) + + assert 'example.com' in certbot.freeze_issue + assert 'test.com' in certbot.freeze_issue + assert certbot.freeze_issue['example.com'] == 10 + assert certbot.freeze_issue['test.com'] == 10 + + def test_find_missing_certificates_skips_ok(self): + """Test that OK certificates are not frozen""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_RETRY_COUNT': '10', + }, clear=False): + certbot = Certbot("/tmp/certs") + + with patch.object(certbot, 'get_certificate_status', return_value='ok'): + certbot.find_missing_certificates(['-d example.com']) + + assert 'example.com' not in certbot.freeze_issue + + @patch('functions.Functions.run_bash') + def test_frozen_host_is_skipped(self, mock_run_bash): + """Test that frozen hosts are skipped during retry period""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_RETRY_COUNT': '2', + }, clear=False): + certbot = Certbot("/tmp/certs") + + # Manually set freeze + certbot.freeze_issue['frozen.com'] = 2 + + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + result = certbot.check_certificates(['frozen.com', 'normal.com']) + + # Should only request certificate for normal.com + call_args = mock_run_bash.call_args[0] + command = call_args[1] + + assert '-d normal.com' in command + assert '-d frozen.com' not in command + # Freeze count should decrement + assert certbot.freeze_issue['frozen.com'] == 1 + + @patch('functions.Functions.run_bash') + def test_frozen_host_unfreezes_after_countdown(self, mock_run_bash): + """Test that frozen hosts are unfrozen after countdown reaches 0""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + certbot = Certbot("/tmp/certs") + + # Set freeze to 1 (will decrement to 0) + certbot.freeze_issue['example.com'] = 1 + + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + # First call: still frozen (count = 1 -> 0) + certbot.check_certificates(['example.com']) + assert certbot.freeze_issue['example.com'] == 0 + + # Second call: should be unfrozen and removed from dict + certbot.check_certificates(['example.com']) + assert 'example.com' not in certbot.freeze_issue + + # Third call: should request certificate + certbot.check_certificates(['example.com']) + + # On third call, certificate should be requested + call_args = mock_run_bash.call_args[0] + command = call_args[1] + assert '-d example.com' in command + + +class TestCertbotWebhookDNS: + """Test manual auth hook (webhook) for DNS challenges""" + + @patch('functions.Functions.run_bash') + def test_dns_challenge_without_webhook(self, mock_run_bash): + """Test DNS challenge command generation without webhook (will fail in practice)""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_PREFERRED_CHALLENGES': 'dns', + }, clear=False): + certbot = Certbot("/tmp/certs") + + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + certbot.check_certificates(['example.com']) + + call_args = mock_run_bash.call_args[0] + command = call_args[1] + + # Should use DNS challenge + assert '--preferred-challenges dns' in command + # Should NOT include HTTP-specific flags + assert '--http-01-port' not in command + assert '--standalone' not in command + # Should NOT include manual flags (no webhook configured) + assert '--manual' not in command + + @patch('functions.Functions.run_bash') + def test_dns_challenge_with_webhook(self, mock_run_bash): + """Test DNS challenge with webhook for wildcard certificates""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_PREFERRED_CHALLENGES': 'dns', + 'EASYHAPROXY_CERTBOT_MANUAL_AUTH_HOOK': '/usr/local/bin/cloudflare-dns.sh', + }, clear=False): + certbot = Certbot("/tmp/certs") + + # DNS is required for wildcard certificates + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + certbot.check_certificates(['*.example.com', 'example.com']) + + call_args = mock_run_bash.call_args[0] + command = call_args[1] + + # Verify DNS challenge with webhook + assert '--preferred-challenges dns' in command + assert '--manual' in command + assert "--manual-auth-hook '/usr/local/bin/cloudflare-dns.sh'" in command + # Verify both wildcard and apex domain + assert '-d *.example.com' in command + assert '-d example.com' in command + + @patch('functions.Functions.run_bash') + def test_http_challenge_with_webhook(self, mock_run_bash): + """Test HTTP challenge can also use webhook (less common)""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_PREFERRED_CHALLENGES': 'http', + 'EASYHAPROXY_CERTBOT_MANUAL_AUTH_HOOK': '/hooks/http-webroot.sh', + }, clear=False): + certbot = Certbot("/tmp/certs") + + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + certbot.check_certificates(['example.com']) + + call_args = mock_run_bash.call_args[0] + command = call_args[1] + + # Both HTTP flags and webhook should be present + assert '--preferred-challenges http' in command + assert '--http-01-port 2080' in command + assert '--standalone' in command + assert '--manual' in command + assert "--manual-auth-hook '/hooks/http-webroot.sh'" in command + + @patch('functions.Functions.run_bash') + def test_webhook_environment_variables_documented(self, mock_run_bash): + """Document environment variables passed to webhook by certbot""" + mock_run_bash.return_value = (0, []) + + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_PREFERRED_CHALLENGES': 'dns', + 'EASYHAPROXY_CERTBOT_MANUAL_AUTH_HOOK': '/hooks/dns-hook.sh', + }, clear=False): + certbot = Certbot("/tmp/certs") + + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + certbot.check_certificates(['example.com']) + + # Certbot automatically passes these to the webhook script: + # CERTBOT_DOMAIN - Domain being authenticated (e.g., "example.com") + # CERTBOT_VALIDATION - Validation string to add to DNS TXT record + # CERTBOT_TOKEN - Challenge token (for HTTP challenges) + # + # Example webhook script: + # #!/bin/bash + # # Add TXT record: _acme-challenge.$CERTBOT_DOMAIN -> $CERTBOT_VALIDATION + # curl -X POST "https://api.cloudflare.com/client/v4/zones/$ZONE_ID/dns_records" \ + # -H "Authorization: Bearer $CF_TOKEN" \ + # -d '{"type":"TXT","name":"_acme-challenge.'$CERTBOT_DOMAIN'","content":"'$CERTBOT_VALIDATION'"}' + + assert mock_run_bash.called + + @patch('functions.Functions.run_bash') + def test_webhook_with_multiple_providers(self, mock_run_bash): + """Test webhook works with different ACME providers""" + mock_run_bash.return_value = (0, []) + + # ZeroSSL with DNS challenge and webhook + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + 'EASYHAPROXY_CERTBOT_SERVER': 'https://acme.zerossl.com/v2/DV90', + 'EASYHAPROXY_CERTBOT_EAB_KID': 'zerossl-kid', + 'EASYHAPROXY_CERTBOT_EAB_HMAC_KEY': 'zerossl-hmac', + 'EASYHAPROXY_CERTBOT_PREFERRED_CHALLENGES': 'dns', + 'EASYHAPROXY_CERTBOT_MANUAL_AUTH_HOOK': '/hooks/route53-dns.py', + }, clear=False): + certbot = Certbot("/tmp/certs") + + with patch.object(certbot, 'get_certificate_status', return_value='not_found'): + with patch.object(certbot, 'find_live_certificates'): + certbot.check_certificates(['example.com']) + + call_args = mock_run_bash.call_args[0] + command = call_args[1] + + # All components should be present + assert '--server https://acme.zerossl.com/v2/DV90' in command + assert '--eab-kid "zerossl-kid"' in command + assert '--eab-hmac-key "zerossl-hmac"' in command + assert '--preferred-challenges dns' in command + assert "--manual-auth-hook '/hooks/route53-dns.py'" in command \ No newline at end of file diff --git a/tests/test_certbot_haproxy_config.py b/tests/test_certbot_haproxy_config.py new file mode 100644 index 0000000..299cb3b --- /dev/null +++ b/tests/test_certbot_haproxy_config.py @@ -0,0 +1,414 @@ +""" +Integration tests for Certbot/ACME HAProxy configuration generation + +Tests that verify the HAProxy configuration is correctly generated for HTTP-01 challenges: +- ACLs for /.well-known/acme-challenge/ paths +- certbot_backend routing to 127.0.0.1:2080 +- ACME challenges bypass SSL redirect +- Multiple domains with certbot enabled +""" + +import os +import sys +from unittest.mock import patch + +# Add src to path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from easymapping import HaproxyConfigGenerator +from functions import ContainerEnv + + +class TestCertbotHAProxyConfig: + """Test HAProxy configuration generation for ACME/certbot""" + + def test_certbot_backend_always_created(self): + """Test that certbot_backend is always present in HAProxy config""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + # Empty config should still have certbot_backend + haproxy_config = cfg.generate({}) + + assert 'backend certbot_backend' in haproxy_config + assert 'server certbot 127.0.0.1:2080' in haproxy_config + + def test_certbot_acl_for_single_domain(self): + """Test ACME challenge ACL for single domain with certbot enabled""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + parsed_object = { + 'container1': { + 'easyhaproxy.http.host': 'example.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '3000', + 'easyhaproxy.http.certbot': 'true', + } + } + + haproxy_config = cfg.generate(parsed_object) + + # Verify ACME challenge ACL + assert 'acl is_certbot_example_com_80 path_beg /.well-known/acme-challenge/' in haproxy_config + + # Verify routing to certbot_backend + assert 'use_backend certbot_backend if is_certbot_example_com_80' in haproxy_config + + # Verify certbot_backend exists + assert 'backend certbot_backend' in haproxy_config + assert 'server certbot 127.0.0.1:2080' in haproxy_config + + def test_certbot_acl_for_multiple_domains(self): + """Test ACME challenge ACLs for multiple domains with certbot enabled""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + parsed_object = { + 'container1': { + 'easyhaproxy.http.host': 'example.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '3000', + 'easyhaproxy.http.certbot': 'true', + }, + 'container2': { + 'easyhaproxy.http.host': 'test.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '4000', + 'easyhaproxy.http.certbot': 'true', + }, + 'container3': { + 'easyhaproxy.http.host': 'nocert.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '5000', + # certbot not enabled + } + } + + haproxy_config = cfg.generate(parsed_object) + + # Verify ACLs for domains with certbot=true + assert 'acl is_certbot_example_com_80 path_beg /.well-known/acme-challenge/' in haproxy_config + assert 'acl is_certbot_test_com_80 path_beg /.well-known/acme-challenge/' in haproxy_config + + # Verify NO ACL for domain without certbot + assert 'acl is_certbot_nocert_com_80' not in haproxy_config + + # Verify routing for each certbot-enabled domain + assert 'use_backend certbot_backend if is_certbot_example_com_80' in haproxy_config + assert 'use_backend certbot_backend if is_certbot_test_com_80' in haproxy_config + + # Verify certbot_backend definition exists (only once) + # Count lines starting with "backend certbot_backend" (not use_backend lines) + backend_lines = [line for line in haproxy_config.split('\n') if line.startswith('backend certbot_backend')] + assert len(backend_lines) == 1 + assert haproxy_config.count('server certbot 127.0.0.1:2080') == 1 + + def test_certbot_bypasses_ssl_redirect(self): + """Test that ACME challenges bypass SSL redirect""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + parsed_object = { + 'container1': { + 'easyhaproxy.http.host': 'example.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '3000', + 'easyhaproxy.http.certbot': 'true', + 'easyhaproxy.http.redirect_ssl': 'true', # Force HTTPS + } + } + + haproxy_config = cfg.generate(parsed_object) + + # Find the redirect rule + lines = haproxy_config.split('\n') + redirect_line = None + for line in lines: + if 'http-request redirect scheme https' in line and 'example_com' in line: + redirect_line = line + break + + assert redirect_line is not None, "SSL redirect rule not found" + + # Verify ACME challenge is excluded from redirect + # Should contain: if !is_certbot_example_com_80 is_rule_... + assert '!is_certbot_example_com_80' in redirect_line + + def test_certbot_with_ssl_clone(self): + """Test certbot with clone_to_ssl (auto-create port 443)""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + parsed_object = { + 'container1': { + 'easyhaproxy.http.host': 'example.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '3000', + 'easyhaproxy.http.certbot': 'true', + 'easyhaproxy.http.clone_to_ssl': 'true', + } + } + + haproxy_config = cfg.generate(parsed_object) + + # Should have HTTP frontend (port 80) with certbot ACL + assert 'frontend http_in_80' in haproxy_config + assert 'acl is_certbot_example_com_80 path_beg /.well-known/acme-challenge/' in haproxy_config + + # Should have HTTPS frontend (port 443) without certbot ACL + # (ACME challenges only happen on HTTP port 80) + assert 'frontend http_in_443' in haproxy_config or 'frontend https_in_443' in haproxy_config + + # Port 443 should NOT have certbot ACL + lines = haproxy_config.split('\n') + in_443_frontend = False + for line in lines: + if 'frontend http_in_443' in line or 'frontend https_in_443' in line: + in_443_frontend = True + if in_443_frontend and 'frontend' in line and '443' not in line: + break # Moved to next frontend + if in_443_frontend and 'is_certbot' in line: + assert False, "ACME challenge ACL should not be in port 443 frontend" + + def test_certbot_without_email_no_acl(self): + """Test that no ACME ACLs are created when email is not configured""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': '', # No email + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + parsed_object = { + 'container1': { + 'easyhaproxy.http.host': 'example.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '3000', + 'easyhaproxy.http.certbot': 'true', # Set but won't work without email + } + } + + haproxy_config = cfg.generate(parsed_object) + + # Should NOT create ACME ACL without email + assert 'acl is_certbot_example_com_80' not in haproxy_config + assert 'use_backend certbot_backend' not in haproxy_config + + # certbot_backend should still exist (always created) + assert 'backend certbot_backend' in haproxy_config + + def test_certbot_acl_naming_special_chars(self): + """Test ACME ACL naming with domains containing special characters""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + parsed_object = { + 'container1': { + 'easyhaproxy.http.host': 'sub-domain.example.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '3000', + 'easyhaproxy.http.certbot': 'true', + } + } + + haproxy_config = cfg.generate(parsed_object) + + # Domain with dots should have them replaced with underscores in ACL name + assert 'acl is_certbot_sub-domain_example_com_80' in haproxy_config + assert 'use_backend certbot_backend if is_certbot_sub-domain_example_com_80' in haproxy_config + + def test_certbot_get_certbot_hosts(self): + """Test that get_certbot_hosts returns list of domains with certbot enabled""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + parsed_object = { + 'container1': { + 'easyhaproxy.http.host': 'example.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '3000', + 'easyhaproxy.http.certbot': 'true', + }, + 'container2': { + 'easyhaproxy.http.host': 'test.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '4000', + 'easyhaproxy.http.certbot': 'true', + }, + 'container3': { + 'easyhaproxy.http.host': 'nocert.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '5000', + } + } + + cfg.generate(parsed_object) + + # Should return list of hosts with certbot=true + certbot_hosts = cfg.certbot_hosts + assert 'example.com' in certbot_hosts + assert 'test.com' in certbot_hosts + assert 'nocert.com' not in certbot_hosts + + def test_certbot_port_must_be_80(self): + """Test that certbot only works on port 80 (HTTP-01 requirement)""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + # Try certbot on port 8080 (not standard HTTP port) + parsed_object = { + 'container1': { + 'easyhaproxy.http.host': 'example.com', + 'easyhaproxy.http.port': '8080', # Non-standard port + 'easyhaproxy.http.localport': '3000', + 'easyhaproxy.http.certbot': 'true', + } + } + + haproxy_config = cfg.generate(parsed_object) + + # ACME ACL should still be created (up to user to ensure proper routing) + # Note: The actual ACME validation will fail if port 80 isn't accessible + assert 'acl is_certbot_example_com_8080' in haproxy_config + + +class TestCertbotHAProxyConfigEdgeCases: + """Test edge cases and error conditions""" + + def test_multiple_containers_same_domain_with_certbot(self): + """Test multiple containers serving the same domain with certbot""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + parsed_object = { + 'container1': { + 'easyhaproxy.http.host': 'example.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '3000', + 'easyhaproxy.http.certbot': 'true', + }, + 'container2': { + 'easyhaproxy.http.host': 'example.com', # Same domain + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '4000', + 'easyhaproxy.http.certbot': 'true', + } + } + + haproxy_config = cfg.generate(parsed_object) + + # Should only create ACL once (not duplicate) + assert haproxy_config.count('acl is_certbot_example_com_80') == 1 + + # Should route to certbot_backend + assert 'use_backend certbot_backend if is_certbot_example_com_80' in haproxy_config + + def test_certbot_with_custom_ports(self): + """Test certbot behavior with custom frontend ports""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + parsed_object = { + 'container1': { + 'easyhaproxy.custom.host': 'example.com', + 'easyhaproxy.custom.port': '8080', + 'easyhaproxy.custom.localport': '3000', + 'easyhaproxy.custom.certbot': 'true', + } + } + + haproxy_config = cfg.generate(parsed_object) + + # Should create frontend on port 8080 with certbot ACL + assert 'frontend http_in_8080' in haproxy_config + assert 'acl is_certbot_example_com_8080' in haproxy_config + + def test_certbot_backend_format(self): + """Test exact format of certbot_backend""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + haproxy_config = cfg.generate({}) + + # Verify exact backend format + assert 'backend certbot_backend' in haproxy_config + assert 'mode http' in haproxy_config + assert 'server certbot 127.0.0.1:2080' in haproxy_config + + # Should NOT have any load balancing, health checks, etc. + # (it's a simple pass-through to localhost) + lines = haproxy_config.split('\n') + in_certbot_backend = False + certbot_backend_lines = [] + for line in lines: + if 'backend certbot_backend' in line: + in_certbot_backend = True + elif in_certbot_backend and line.strip() and not line.startswith(' '): + break # End of backend section + elif in_certbot_backend: + certbot_backend_lines.append(line.strip()) + + # Should only have mode and server lines + assert 'mode http' in certbot_backend_lines + assert 'server certbot 127.0.0.1:2080' in certbot_backend_lines + assert len([l for l in certbot_backend_lines if l]) == 2 # Only 2 non-empty lines + + def test_certbot_acl_order_before_use_backend(self): + """Test that ACL definitions come before use_backend rules""" + with patch.dict(os.environ, { + 'EASYHAPROXY_CERTBOT_EMAIL': 'test@example.com', + }, clear=False): + mapping = ContainerEnv.read() + cfg = HaproxyConfigGenerator(mapping) + + parsed_object = { + 'container1': { + 'easyhaproxy.http.host': 'example.com', + 'easyhaproxy.http.port': '80', + 'easyhaproxy.http.localport': '3000', + 'easyhaproxy.http.certbot': 'true', + } + } + + haproxy_config = cfg.generate(parsed_object) + + # Find positions + acl_pos = haproxy_config.find('acl is_certbot_example_com_80') + use_backend_pos = haproxy_config.find('use_backend certbot_backend if is_certbot_example_com_80') + + assert acl_pos > 0, "ACL not found" + assert use_backend_pos > 0, "use_backend not found" + assert acl_pos < use_backend_pos, "ACL must be defined before use_backend" \ No newline at end of file