movefasta/nixpkgs
0
0
Fork 0
mirror of https://github.com/NixOS/nixpkgs.git synced 2025-07-14 14:10:33 +03:00
Code Activity

nixos/tests/acme: Refactor test suite

Browse source 
Split tests up based on certain use cases:

- http01-builtin: Tests most functionality of the core module, such
 as the systemd and hashing components, whilst utilising lego's built
 in http01 resolution mechanis.
- dns01: Tests only that DNS01 renewal works as expected.
- nginx: Tests nginx compatability
- httpd: Tests httpd compatability
- caddy: Tests caddy compatability
This commit is contained in:
Lucas Savva 2025-01-20 16:38:12 +00:00
parent 84af416af6
commit 229640ed3a
19 changed files with 944 additions and 1061 deletions
Download patch file Download diff file Expand all files Collapse all files

32
nixos/modules/security/acme/default.nix
View file

@ -117,6 +117,7 @@ let
${lib.optionalString cfg.defaults.enableDebugLogs "set -x"}
set -euo pipefail
cd /var/lib/acme
chmod -R u=rwX,g=,o= .lego/accounts
chown -R ${user} .lego/accounts
'' + (lib.concatStringsSep "\n" (lib.mapAttrsToList (cert: data: ''
for fixpath in ${lib.escapeShellArg cert} .lego/${lib.escapeShellArg cert}; do
@ -161,8 +162,7 @@ let
};
}
# In order to avoid race conditions creating the CA for selfsigned certs,
# we have a separate service which will create the necessary files.
# Avoid race conditions creating the CA for selfsigned certs
(lib.mkIf cfg.preliminarySelfsigned {
path = [ pkgs.minica ];
# Working directory will be /tmp
@ -482,6 +482,9 @@ let
# By default group will have no access to the cert files.
# This chmod will fix that.
chmod 640 out/*
# Also ensure safer permissions on the account directory.
chmod -R u=rwX,g=,o= accounts/.
'';
};
};
@ -599,6 +602,17 @@ let
'';
};
listenHTTP = lib.mkOption {
type = lib.types.nullOr lib.types.str;
inherit (defaultAndText "listenHTTP" null) default defaultText;
example = ":1360";
description = ''
Interface and port to listen on to solve HTTP challenges
in the form `[INTERFACE]:PORT`.
If you use a port other than 80, you must proxy port 80 to this port.
'';
};
dnsProvider = lib.mkOption {
type = lib.types.nullOr lib.types.str;
inherit (defaultAndText "dnsProvider" null) default defaultText;
@ -744,20 +758,6 @@ let
'';
};
# This setting must be different for each configured certificate, otherwise
# two or more renewals may fail to bind to the address. Hence, it is not in
# the inheritableOpts.
listenHTTP = lib.mkOption {
type = lib.types.nullOr lib.types.str;
default = null;
example = ":1360";
description = ''
Interface and port to listen on to solve HTTP challenges
in the form [INTERFACE]:PORT.
If you use a port other than 80, you must proxy port 80 to this port.
'';
};
s3Bucket = lib.mkOption {
type = lib.types.nullOr lib.types.str;
default = null;

3
nixos/release-combined.nix
View file

@ -75,7 +75,8 @@ rec {
(onFullSupported "nixos.iso_gnome")
(onFullSupported "nixos.manual")
(onSystems [ "aarch64-linux" ] "nixos.sd_image")
(onFullSupported "nixos.tests.acme")
(onFullSupported "nixos.tests.acme.http01-builtin")
(onFullSupported "nixos.tests.acme.dns01")
(onSystems [ "x86_64-linux" ] "nixos.tests.boot.biosCdrom")
(onSystems [ "x86_64-linux" ] "nixos.tests.boot.biosUsb")
(onFullSupported "nixos.tests.boot-stage1")

8
nixos/release-small.nix
View file

@ -48,8 +48,11 @@ rec {
dummy
;
tests = {
inherit (nixos'.tests.acme)
http01-builtin
dns01
;
inherit (nixos'.tests)
acme
containers-imperative
containers-ip
firewall
@ -135,7 +138,8 @@ rec {
(map onSupported [
"nixos.dummy"
"nixos.manual"
"nixos.tests.acme"
"nixos.tests.acme.http01-builtin"
"nixos.tests.acme.dns01"
"nixos.tests.containers-imperative"
"nixos.tests.containers-ip"
"nixos.tests.firewall"

788
nixos/tests/acme.nix
View file

@ -1,788 +0,0 @@
{ config, lib, ... }: let
pkgs = config.node.pkgs;
commonConfig = ./common/acme/client;
dnsServerIP = nodes: nodes.dnsserver.networking.primaryIPAddress;
dnsScript = nodes: let
dnsAddress = dnsServerIP nodes;
in pkgs.writeShellScript "dns-hook.sh" ''
set -euo pipefail
echo '[INFO]' "[$2]" 'dns-hook.sh' $*
if [ "$1" = "present" ]; then
${pkgs.curl}/bin/curl --data '{"host": "'"$2"'", "value": "'"$3"'"}' http://${dnsAddress}:8055/set-txt
else
${pkgs.curl}/bin/curl --data '{"host": "'"$2"'"}' http://${dnsAddress}:8055/clear-txt
fi
'';
dnsConfig = nodes: {
dnsProvider = "exec";
dnsPropagationCheck = false;
environmentFile = pkgs.writeText "wildcard.env" ''
EXEC_PATH=${dnsScript nodes}
EXEC_POLLING_INTERVAL=1
EXEC_PROPAGATION_TIMEOUT=1
EXEC_SEQUENCE_INTERVAL=1
'';
};
documentRoot = pkgs.runCommand "docroot" {} ''
mkdir -p "$out"
echo hello world > "$out/index.html"
'';
vhostBase = {
forceSSL = true;
locations."/".root = documentRoot;
};
vhostBaseHttpd = {
forceSSL = true;
inherit documentRoot;
};
simpleConfig = {
security.acme = {
certs."http.example.test" = {
listenHTTP = ":80";
};
};
networking.firewall.allowedTCPPorts = [ 80 ];
};
# Base specialisation config for testing general ACME features
webserverBasicConfig = {
services.nginx.enable = true;
services.nginx.virtualHosts."a.example.test" = vhostBase // {
enableACME = true;
};
};
# Generate specialisations for testing a web server
mkServerConfigs = { server, group, vhostBaseData, extraConfig ? {} }: let
baseConfig = { nodes, config, specialConfig ? {} }: lib.mkMerge [
{
security.acme = {
defaults = (dnsConfig nodes);
# One manual wildcard cert
certs."example.test" = {
domain = "*.example.test";
};
};
users.users."${config.services."${server}".user}".extraGroups = ["acme"];
services."${server}" = {
enable = true;
virtualHosts = {
# Run-of-the-mill vhost using HTTP-01 validation
"${server}-http.example.test" = vhostBaseData // {
serverAliases = [ "${server}-http-alias.example.test" ];
enableACME = true;
};
# Another which inherits the DNS-01 config
"${server}-dns.example.test" = vhostBaseData // {
serverAliases = [ "${server}-dns-alias.example.test" ];
enableACME = true;
# Set acmeRoot to null instead of using the default of "/var/lib/acme/acme-challenge"
# webroot + dnsProvider are mutually exclusive.
acmeRoot = null;
};
# One using the wildcard certificate
"${server}-wildcard.example.test" = vhostBaseData // {
serverAliases = [ "${server}-wildcard-alias.example.test" ];
useACMEHost = "example.test";
};
} // (lib.optionalAttrs (server == "nginx") {
# The nginx module supports using a different key than the hostname
different-key = vhostBaseData // {
serverName = "${server}-different-key.example.test";
serverAliases = [ "${server}-different-key-alias.example.test" ];
enableACME = true;
};
});
};
# Used to determine if service reload was triggered
systemd.targets."test-renew-${server}" = {
wants = [ "acme-${server}-http.example.test.service" ];
after = [ "acme-${server}-http.example.test.service" "${server}-config-reload.service" ];
};
}
specialConfig
extraConfig
];
in {
"${server}".configuration = { nodes, config, ... }: baseConfig {
inherit nodes config;
};
# Test that server reloads when an alias is removed (and subsequently test removal works in acme)
"${server}_remove_alias".configuration = { nodes, config, ... }: baseConfig {
inherit nodes config;
specialConfig = {
# Remove an alias, but create a standalone vhost in its place for testing.
# This configuration results in certificate errors as useACMEHost does not imply
# append extraDomains, and thus we can validate the SAN is removed.
services."${server}" = {
virtualHosts."${server}-http.example.test".serverAliases = lib.mkForce [];
virtualHosts."${server}-http-alias.example.test" = vhostBaseData // {
useACMEHost = "${server}-http.example.test";
};
};
};
};
# Test that the server reloads when only the acme configuration is changed.
"${server}_change_acme_conf".configuration = { nodes, config, ... }: baseConfig {
inherit nodes config;
specialConfig = {
security.acme.certs."${server}-http.example.test" = {
keyType = "ec384";
# Also test that postRun is exec'd as root
postRun = "id | grep root";
};
};
};
};
in {
name = "acme";
meta = {
maintainers = lib.teams.acme.members;
# Hard timeout in seconds. Average run time is about 7 minutes.
timeout = 1800;
};
nodes = {
# The fake ACME server which will respond to client requests
acme = { nodes, ... }: {
imports = [ ./common/acme/server ];
networking.nameservers = lib.mkForce [ (dnsServerIP nodes) ];
};
# A fake DNS server which can be configured with records as desired
# Used to test DNS-01 challenge
dnsserver = { nodes, ... }: {
networking.firewall.allowedTCPPorts = [ 8055 53 ];
networking.firewall.allowedUDPPorts = [ 53 ];
systemd.services.pebble-challtestsrv = {
enable = true;
description = "Pebble ACME challenge test server";
wantedBy = [ "network.target" ];
serviceConfig = {
ExecStart = "${pkgs.pebble}/bin/pebble-challtestsrv -dns01 ':53' -defaultIPv6 '' -defaultIPv4 '${nodes.webserver.networking.primaryIPAddress}'";
# Required to bind on privileged ports.
AmbientCapabilities = [ "CAP_NET_BIND_SERVICE" ];
};
};
};
# A web server which will be the node requesting certs
webserver = { nodes, config, ... }: {
imports = [ commonConfig ];
networking.nameservers = lib.mkForce [ (dnsServerIP nodes) ];
networking.firewall.allowedTCPPorts = [ 80 443 ];
# OpenSSL will be used for more thorough certificate validation
environment.systemPackages = [ pkgs.openssl ];
# Set log level to info so that we can see when the service is reloaded
services.nginx.logError = "stderr info";
specialisation = {
# Tests HTTP-01 verification using Lego's built-in web server
http01lego.configuration = simpleConfig;
# account hash generation with default server from <= 23.11
http01lego_legacyAccountHash.configuration = lib.mkMerge [
simpleConfig
{
security.acme.defaults.server = lib.mkForce null;
}
];
renew.configuration = lib.mkMerge [
simpleConfig
{
# Pebble provides 5 year long certs,
# needs to be higher than that to test renewal
security.acme.certs."http.example.test".validMinDays = 9999;
}
];
# Tests that account creds can be safely changed.
accountchange.configuration = lib.mkMerge [
simpleConfig
{
security.acme.certs."http.example.test".email = "admin@example.test";
}
];
# First derivation used to test general ACME features
general.configuration = { ... }: let
caDomain = nodes.acme.test-support.acme.caDomain;
email = config.security.acme.defaults.email;
# Exit 99 to make it easier to track if this is the reason a renew failed
accountCreateTester = ''
test -e accounts/${caDomain}/${email}/account.json || exit 99
'';
in lib.mkMerge [
webserverBasicConfig
{
# Used to test that account creation is collated into one service.
# These should not run until after acme-finished-a.example.test.target
systemd.services."b.example.test".preStart = accountCreateTester;
systemd.services."c.example.test".preStart = accountCreateTester;
services.nginx.virtualHosts."b.example.test" = vhostBase // {
enableACME = true;
};
services.nginx.virtualHosts."c.example.test" = vhostBase // {
enableACME = true;
};
}
];
# Test OCSP Stapling
ocsp_stapling.configuration = { ... }: lib.mkMerge [
webserverBasicConfig
{
security.acme.certs."a.example.test".ocspMustStaple = true;
services.nginx.virtualHosts."a.example.test" = {
extraConfig = ''
ssl_stapling on;
ssl_stapling_verify on;
'';
};
}
];
# Validate service relationships by adding a slow start service to nginx' wants.
# Reproducer for https://github.com/NixOS/nixpkgs/issues/81842
slow_startup.configuration = { ... }: lib.mkMerge [
webserverBasicConfig
{
systemd.services.my-slow-service = {
wantedBy = [ "multi-user.target" "nginx.service" ];
before = [ "nginx.service" ];
preStart = "sleep 5";
script = "${pkgs.python3}/bin/python -m http.server";
};
services.nginx.virtualHosts."slow.example.test" = {
forceSSL = true;
enableACME = true;
locations."/".proxyPass = "http://localhost:8000";
};
}
];
concurrency_limit.configuration = {pkgs, ...}: lib.mkMerge [
webserverBasicConfig {
security.acme.maxConcurrentRenewals = 1;
services.nginx.virtualHosts = {
"f.example.test" = vhostBase // {
enableACME = true;
};
"g.example.test" = vhostBase // {
enableACME = true;
};
"h.example.test" = vhostBase // {
enableACME = true;
};
};
systemd.services = {
# check for mutual exclusion of starting renew services
"acme-f.example.test".serviceConfig.ExecPreStart = "+" + (pkgs.writeShellScript "test-f" ''
test "$(systemctl is-active acme-{g,h}.example.test.service | grep activating | wc -l)" -le 0
'');
"acme-g.example.test".serviceConfig.ExecPreStart = "+" + (pkgs.writeShellScript "test-g" ''
test "$(systemctl is-active acme-{f,h}.example.test.service | grep activating | wc -l)" -le 0
'');
"acme-h.example.test".serviceConfig.ExecPreStart = "+" + (pkgs.writeShellScript "test-h" ''
test "$(systemctl is-active acme-{g,f}.example.test.service | grep activating | wc -l)" -le 0
'');
};
}
];
# Test lego internal server (listenHTTP option)
# Also tests useRoot option
lego_server.configuration = { ... }: {
security.acme.useRoot = true;
security.acme.certs."lego.example.test" = {
listenHTTP = ":80";
group = "nginx";
};
services.nginx.enable = true;
services.nginx.virtualHosts."lego.example.test" = {
useACMEHost = "lego.example.test";
onlySSL = true;
};
};
# Test compatibility with Caddy
# It only supports useACMEHost, hence not using mkServerConfigs
} // (let
baseCaddyConfig = { nodes, config, ... }: {
security.acme = {
defaults = (dnsConfig nodes);
# One manual wildcard cert
certs."example.test" = {
domain = "*.example.test";
};
};
users.users."${config.services.caddy.user}".extraGroups = ["acme"];
services.caddy = {
enable = true;
virtualHosts."a.example.test" = {
useACMEHost = "example.test";
extraConfig = ''
root * ${documentRoot}
'';
};
};
};
in {
caddy.configuration = baseCaddyConfig;
# Test that the server reloads when only the acme configuration is changed.
"caddy_change_acme_conf".configuration = { nodes, config, ... }: lib.mkMerge [
(baseCaddyConfig {
inherit nodes config;
})
{
security.acme.certs."example.test" = {
keyType = "ec384";
};
}
];
# Test compatibility with Nginx
}) // (mkServerConfigs {
server = "nginx";
group = "nginx";
vhostBaseData = vhostBase;
})
# Test compatibility with Apache HTTPD
// (mkServerConfigs {
server = "httpd";
group = "wwwrun";
vhostBaseData = vhostBaseHttpd;
extraConfig = {
services.httpd.adminAddr = config.security.acme.defaults.email;
};
});
};
# The client will be used to curl the webserver to validate configuration
client = { nodes, ... }: {
imports = [ commonConfig ];
networking.nameservers = lib.mkForce [ (dnsServerIP nodes) ];
# OpenSSL will be used for more thorough certificate validation
environment.systemPackages = [ pkgs.openssl ];
};
};
testScript = { nodes, ... }:
let
caDomain = nodes.acme.test-support.acme.caDomain;
in
# Note, wait_for_unit does not work for oneshot services that do not have RemainAfterExit=true,
# this is because a oneshot goes from inactive => activating => inactive, and never
# reaches the active state. Targets do not have this issue.
''
import time
TOTAL_RETRIES = 20
class BackoffTracker(object):
delay = 1
increment = 1
def handle_fail(self, retries, message) -> int:
assert retries < TOTAL_RETRIES, message
print(f"Retrying in {self.delay}s, {retries + 1}/{TOTAL_RETRIES}")
time.sleep(self.delay)
# Only increment after the first try
if retries == 0:
self.delay += self.increment
self.increment *= 2
return retries + 1
def protect(self, func):
def wrapper(*args, retries: int = 0, **kwargs):
try:
return func(*args, **kwargs)
except Exception as err:
retries = self.handle_fail(retries, err.args)
return wrapper(*args, retries=retries, **kwargs)
return wrapper
backoff = BackoffTracker()
def switch_to(node, name, allow_fail=False):
# On first switch, this will create a symlink to the current system so that we can
# quickly switch between derivations
root_specs = "/tmp/specialisation"
node.execute(
f"test -e {root_specs}"
f" || ln -s $(readlink /run/current-system)/specialisation {root_specs}"
)
switcher_path = (
f"/run/current-system/specialisation/{name}/bin/switch-to-configuration"
)
rc, _ = node.execute(f"test -e '{switcher_path}'")
if rc > 0:
switcher_path = f"/tmp/specialisation/{name}/bin/switch-to-configuration"
if not allow_fail:
node.succeed(
f"{switcher_path} test"
)
else:
node.execute(
f"{switcher_path} test"
)
# Start a unit explicitly, then wait for it to activate.
# This is used for the acme-finished-* targets, as those
# aren't started by switch-to-configuration, meaning
# wait_for_unit(target) will fail with "no pending jobs"
# if it wins the race and checks the target state before
# the actual unit is started.
def start_and_wait(node, unit):
node.start_job(unit)
node.wait_for_unit(unit)
# Ensures the issuer of our cert matches the chain
# and matches the issuer we expect it to be.
# It's a good validation to ensure the cert.pem and fullchain.pem
# are not still selfsigned after verification
def check_issuer(node, cert_name, issuer):
for fname in ("cert.pem", "fullchain.pem"):
actual_issuer = node.succeed(
f"openssl x509 -noout -issuer -in /var/lib/acme/{cert_name}/{fname}"
).partition("=")[2]
assert (
issuer.lower() in actual_issuer.lower()
), f"{fname} issuer mismatch. Expected {issuer} got {actual_issuer}"
# Ensure cert comes before chain in fullchain.pem
def check_fullchain(node, cert_name):
cert_file = f"/var/lib/acme/{cert_name}/fullchain.pem"
num_certs = node.succeed(f"grep -o 'END CERTIFICATE' {cert_file}")
assert len(num_certs.strip().split("\n")) > 1, "Insufficient certs in fullchain.pem"
first_cert_data = node.succeed(
f"grep -m1 -B50 'END CERTIFICATE' {cert_file}"
" | openssl x509 -noout -text"
)
for line in first_cert_data.lower().split("\n"):
if "dns:" in line:
print(f"First DNSName in fullchain.pem: {line}")
assert cert_name.lower() in line, f"{cert_name} not found in {line}"
return
assert False
@backoff.protect
def check_connection(node, domain):
result = node.succeed(
"openssl s_client -brief -verify 2 -CAfile /tmp/ca.crt"
f" -servername {domain} -connect {domain}:443 < /dev/null 2>&1"
)
for line in result.lower().split("\n"):
assert not (
"verification" in line and "error" in line
), f"Failed to connect to https://{domain}"
@backoff.protect
def check_connection_key_bits(node, domain, bits):
result = node.succeed(
"openssl s_client -CAfile /tmp/ca.crt"
f" -servername {domain} -connect {domain}:443 < /dev/null"
" | openssl x509 -noout -text | grep -i Public-Key"
)
print("Key type:", result)
assert bits in result, f"Did not find expected number of bits ({bits}) in key"
@backoff.protect
def check_stapling(node, domain):
# Pebble doesn't provide a full OCSP responder, so just check the URL
result = node.succeed(
"openssl s_client -CAfile /tmp/ca.crt"
f" -servername {domain} -connect {domain}:443 < /dev/null"
" | openssl x509 -noout -ocsp_uri"
)
print("OCSP Responder URL:", result)
assert "${caDomain}:4002" in result.lower(), "OCSP Stapling check failed"
@backoff.protect
def download_ca_certs(node):
node.succeed("curl https://${caDomain}:15000/roots/0 > /tmp/ca.crt")
node.succeed("curl https://${caDomain}:15000/intermediate-keys/0 >> /tmp/ca.crt")
@backoff.protect
def set_a_record(node):
node.succeed(
'curl --data \'{"host": "${caDomain}", "addresses": ["${nodes.acme.networking.primaryIPAddress}"]}\' http://${dnsServerIP nodes}:8055/add-a'
)
start_all()
dnsserver.wait_for_unit("pebble-challtestsrv.service")
client.wait_for_unit("default.target")
set_a_record(client)
acme.systemctl("start network-online.target")
acme.wait_for_unit("network-online.target")
acme.wait_for_unit("pebble.service")
download_ca_certs(client)
# Perform http-01 w/ lego test first
with subtest("Can request certificate with Lego's built in web server"):
switch_to(webserver, "http01lego")
start_and_wait(webserver, "acme-finished-http.example.test.target")
check_fullchain(webserver, "http.example.test")
check_issuer(webserver, "http.example.test", "pebble")
# Perform account hash test
with subtest("Assert that account hash didn't unexpectedly change"):
hash = webserver.succeed("ls /var/lib/acme/.lego/accounts/")
print("Account hash: " + hash)
assert hash.strip() == "d590213ed52603e9128d"
# Perform renewal test
with subtest("Can renew certificates when they expire"):
hash = webserver.succeed("sha256sum /var/lib/acme/http.example.test/cert.pem")
switch_to(webserver, "renew")
start_and_wait(webserver, "acme-finished-http.example.test.target")
check_fullchain(webserver, "http.example.test")
check_issuer(webserver, "http.example.test", "pebble")
hash_after = webserver.succeed("sha256sum /var/lib/acme/http.example.test/cert.pem")
assert hash != hash_after
# Perform account change test
with subtest("Handles email change correctly"):
hash = webserver.succeed("sha256sum /var/lib/acme/http.example.test/cert.pem")
switch_to(webserver, "accountchange")
start_and_wait(webserver, "acme-finished-http.example.test.target")
check_fullchain(webserver, "http.example.test")
check_issuer(webserver, "http.example.test", "pebble")
hash_after = webserver.succeed("sha256sum /var/lib/acme/http.example.test/cert.pem")
# Has to do a full run to register account, which creates new certs.
assert hash != hash_after
# Perform general tests
switch_to(webserver, "general")
with subtest("Can request certificate with HTTP-01 challenge"):
start_and_wait(webserver, "acme-finished-a.example.test.target")
check_fullchain(webserver, "a.example.test")
check_issuer(webserver, "a.example.test", "pebble")
webserver.wait_for_unit("nginx.service")
check_connection(client, "a.example.test")
with subtest("Runs 1 cert for account creation before others"):
start_and_wait(webserver, "acme-finished-b.example.test.target")
start_and_wait(webserver, "acme-finished-c.example.test.target")
check_connection(client, "b.example.test")
check_connection(client, "c.example.test")
with subtest("Certificates and accounts have safe + valid permissions"):
# Nginx will set the group appropriately when enableACME is used
group = "nginx"
webserver.succeed(
f"test $(stat -L -c '%a %U %G' /var/lib/acme/a.example.test/*.pem | tee /dev/stderr | grep '640 acme {group}' | wc -l) -eq 5"
)
webserver.succeed(
f"test $(stat -L -c '%a %U %G' /var/lib/acme/.lego/a.example.test/**/a.example.test* | tee /dev/stderr | grep '600 acme {group}' | wc -l) -eq 4"
)
webserver.succeed(
f"test $(stat -L -c '%a %U %G' /var/lib/acme/a.example.test | tee /dev/stderr | grep '750 acme {group}' | wc -l) -eq 1"
)
webserver.succeed(
f"test $(find /var/lib/acme/accounts -type f -exec stat -L -c '%a %U %G' {{}} \\; | tee /dev/stderr | grep -v '600 acme {group}' | wc -l) -eq 0"
)
# Selfsigned certs tests happen late so we aren't fighting the system init triggering cert renewal
with subtest("Can generate valid selfsigned certs"):
webserver.succeed("systemctl clean acme-a.example.test.service --what=state")
webserver.succeed("systemctl start acme-selfsigned-a.example.test.service")
check_fullchain(webserver, "a.example.test")
check_issuer(webserver, "a.example.test", "minica")
# Check selfsigned permissions
webserver.succeed(
f"test $(stat -L -c '%a %U %G' /var/lib/acme/a.example.test/*.pem | tee /dev/stderr | grep '640 acme {group}' | wc -l) -eq 5"
)
# Will succeed if nginx can load the certs
webserver.succeed("systemctl start nginx-config-reload.service")
with subtest("Correctly implements OCSP stapling"):
switch_to(webserver, "ocsp_stapling")
start_and_wait(webserver, "acme-finished-a.example.test.target")
check_stapling(client, "a.example.test")
with subtest("Can request certificate with HTTP-01 using lego's internal web server"):
switch_to(webserver, "lego_server")
start_and_wait(webserver, "acme-finished-lego.example.test.target")
webserver.wait_for_unit("nginx.service")
webserver.succeed("echo HENLO && systemctl cat nginx.service")
webserver.succeed('test "$(stat -c \'%U\' /var/lib/acme/* | uniq)" = "root"')
check_connection(client, "a.example.test")
check_connection(client, "lego.example.test")
with subtest("Can request certificate with HTTP-01 when nginx startup is delayed"):
webserver.execute("systemctl stop nginx")
switch_to(webserver, "slow_startup")
start_and_wait(webserver, "acme-finished-slow.example.test.target")
check_issuer(webserver, "slow.example.test", "pebble")
webserver.wait_for_unit("nginx.service")
check_connection(client, "slow.example.test")
with subtest("Can limit concurrency of running renewals"):
switch_to(webserver, "concurrency_limit")
start_and_wait(webserver, "acme-finished-f.example.test.target")
start_and_wait(webserver, "acme-finished-g.example.test.target")
start_and_wait(webserver, "acme-finished-h.example.test.target")
check_connection(client, "f.example.test")
check_connection(client, "g.example.test")
check_connection(client, "h.example.test")
with subtest("Works with caddy"):
switch_to(webserver, "caddy")
start_and_wait(webserver, "acme-finished-example.test.target")
webserver.wait_for_unit("caddy.service")
# FIXME reloading caddy is not sufficient to load new certs.
# Restart it manually until this is fixed.
webserver.succeed("systemctl restart caddy.service")
check_connection(client, "a.example.test")
with subtest("security.acme changes reflect on caddy"):
switch_to(webserver, "caddy_change_acme_conf")
start_and_wait(webserver, "acme-finished-example.test.target")
webserver.wait_for_unit("caddy.service")
# FIXME reloading caddy is not sufficient to load new certs.
# Restart it manually until this is fixed.
webserver.succeed("systemctl restart caddy.service")
check_connection_key_bits(client, "a.example.test", "384")
common_domains = ["http", "dns", "wildcard"]
for server, logsrc, domains in [
("nginx", "journalctl -n 30 -u nginx.service", common_domains + ["different-key"]),
("httpd", "tail -n 30 /var/log/httpd/*.log", common_domains),
]:
wait_for_server = lambda: webserver.wait_for_unit(f"{server}.service")
with subtest(f"Works with {server}"):
try:
switch_to(webserver, server)
for domain in domains:
if domain != "wildcard":
start_and_wait(
webserver,
f"acme-finished-{server}-{domain}.example.test.target"
)
except Exception as err:
_, output = webserver.execute(
f"{logsrc} && ls -al /var/lib/acme/acme-challenge"
)
print(output)
raise err
wait_for_server()
for domain in domains:
if domain != "wildcard":
check_issuer(webserver, f"{server}-{domain}.example.test", "pebble")
for domain in domains:
check_connection(client, f"{server}-{domain}.example.test")
check_connection(client, f"{server}-{domain}-alias.example.test")
test_domain = f"{server}-{domains[0]}.example.test"
with subtest(f"Can reload {server} when timer triggers renewal"):
# Switch to selfsigned first
webserver.succeed(f"systemctl clean acme-{test_domain}.service --what=state")
webserver.succeed(f"systemctl start acme-selfsigned-{test_domain}.service")
check_issuer(webserver, test_domain, "minica")
webserver.succeed(f"systemctl start {server}-config-reload.service")
webserver.succeed(f"systemctl start test-renew-{server}.target")
check_issuer(webserver, test_domain, "pebble")
check_connection(client, test_domain)
with subtest("Can remove an alias from a domain + cert is updated"):
test_alias = f"{server}-{domains[0]}-alias.example.test"
switch_to(webserver, f"{server}_remove_alias")
wait_for_server()
start_and_wait(webserver, f"acme-finished-{test_domain}.target")
wait_for_server()
check_connection(client, test_domain)
rc, _s = client.execute(
f"openssl s_client -CAfile /tmp/ca.crt -connect {test_alias}:443"
" </dev/null 2>/dev/null | openssl x509 -noout -text"
f" | grep DNS: | grep {test_alias}"
)
assert rc > 0, "Removed extraDomainName was not removed from the cert"
with subtest("security.acme changes reflect on web server"):
# Switch back to normal server config first, reset everything.
switch_to(webserver, server)
wait_for_server()
switch_to(webserver, f"{server}_change_acme_conf")
start_and_wait(webserver, f"acme-finished-{test_domain}.target")
wait_for_server()
check_connection_key_bits(client, test_domain, "384")
# Perform http-01 w/ lego test again, but using the pre-24.05 account hashing
# (see https://github.com/NixOS/nixpkgs/pull/317257)
with subtest("Check account hashing compatibility with pre-24.05 settings"):
webserver.succeed("rm -rf /var/lib/acme/.lego/accounts/*")
switch_to(webserver, "http01lego_legacyAccountHash", allow_fail=True)
# unit is failed, but in a way that this throws no exception:
try:
start_and_wait(webserver, "acme-finished-http.example.test.target")
except Exception:
# The unit is allowed – or even expected – to fail due to not being able to
# reach the actual letsencrypt server. We only use it for serialising the
# test execution, such that the account check is done after the service run
# involving the account creation has been executed at least once.
pass
hash = webserver.succeed("ls /var/lib/acme/.lego/accounts/")
print("Account hash: " + hash)
assert hash.strip() == "1ccf607d9aa280e9af00"
'';
}

120
nixos/tests/acme/caddy.nix Normal file
View file

@ -0,0 +1,120 @@
{
config,
lib,
pkgs,
...
}:
let
domain = "example.test";
in
{
# Caddy only supports useACMEHost, hence we use a distinct test suite
name = "caddy";
meta = {
maintainers = lib.teams.acme.members;
# Hard timeout in seconds. Average run time is about 60 seconds.
timeout = 180;
};
nodes = {
# The fake ACME server which will respond to client requests
acme =
{ nodes, ... }:
{
imports = [ ../common/acme/server ];
};
caddy =
{ nodes, config, ... }:
let
fqdn = config.networking.fqdn;
in
{
imports = [ ../common/acme/client ];
networking.domain = domain;
networking.firewall.allowedTCPPorts = [
80
443
];
# Resolve the vhosts the easy way
networking.hosts."127.0.0.1" = [
"caddy-alt.${domain}"
];
# OpenSSL will be used for more thorough certificate validation
environment.systemPackages = [ pkgs.openssl ];
security.acme.certs."${fqdn}" = {
listenHTTP = ":8080";
reloadServices = [ "caddy.service" ];
};
users.users."${config.services.caddy.user}".extraGroups = [ "acme" ];
services.caddy = {
enable = true;
# FIXME reloading caddy is not sufficient to load new certs.
# Restart it manually until this is fixed.
enableReload = false;
globalConfig = ''
auto_https off
'';
virtualHosts."${fqdn}:443" = {
useACMEHost = fqdn;
};
virtualHosts.":80".extraConfig = ''
reverse_proxy localhost:8080
'';
};
specialisation.add_domain.configuration = {
security.acme.certs.${fqdn}.extraDomainNames = [
"caddy-alt.${domain}"
];
};
};
};
testScript =
{ nodes, ... }:
''
${(import ./utils.nix).pythonUtils}
domain = "${domain}"
ca_domain = "${nodes.acme.test-support.acme.caDomain}"
fqdn = "${nodes.caddy.networking.fqdn}"
acme.start()
wait_for_running(acme)
acme.wait_for_open_port(443)
with subtest("Boot and acquire a new cert"):
caddy.start()
wait_for_running(caddy)
check_issuer(caddy, fqdn, "pebble")
check_domain(caddy, fqdn, fqdn)
download_ca_certs(caddy, ca_domain)
check_connection(caddy, fqdn)
with subtest("Can run on selfsigned certificates"):
# Switch to selfsigned first
caddy.succeed(f"systemctl clean acme-{fqdn}.service --what=state")
caddy.succeed(f"systemctl start acme-selfsigned-{fqdn}.service")
check_issuer(caddy, fqdn, "minica")
caddy.succeed("systemctl restart caddy.service")
# Check that the web server has picked up the selfsigned cert
check_connection(caddy, fqdn, minica=True)
caddy.succeed(f"systemctl start acme-{fqdn}.service")
# This may fail a couple of times before caddy is restarted
check_issuer(caddy, fqdn, "pebble")
check_connection(caddy, fqdn)
with subtest("security.acme changes reflect on caddy"):
check_connection(caddy, f"caddy-alt.{domain}", fail=True)
switch_to(caddy, "add_domain")
check_connection(caddy, f"caddy-alt.{domain}")
'';
}

56
nixos/tests/acme/default.nix Normal file
View file

@ -0,0 +1,56 @@
{ runTest }:
{
http01-builtin = runTest ./http01-builtin.nix;
dns01 = runTest ./dns01.nix;
caddy = runTest ./caddy.nix;
nginx = runTest (
import ./webserver.nix {
serverName = "nginx";
group = "nginx";
baseModule = {
services.nginx = {
enable = true;
enableReload = true;
logError = "stderr info";
# This tests a number of things at once:
# - Self-signed certs are in place before the webserver startup
# - Nginx is started before acme renewal is attempted
# - useACMEHost behaves as expected
# - acmeFallbackHost behaves as expected
virtualHosts.default = {
default = true;
addSSL = true;
useACMEHost = "proxied.example.test";
acmeFallbackHost = "localhost:8080";
# lego will refuse the request if the host header is not correct
extraConfig = ''
proxy_set_header Host $host;
'';
};
};
};
}
);
httpd = runTest (
import ./webserver.nix {
serverName = "httpd";
group = "wwwrun";
baseModule = {
services.httpd = {
enable = true;
# This is the default by virtue of being the first defined vhost.
virtualHosts.default = {
addSSL = true;
useACMEHost = "proxied.example.test";
locations."/.well-known/acme-challenge" = {
proxyPass = "http://localhost:8080/.well-known/acme-challenge";
extraConfig = ''
ProxyPreserveHost On
'';
};
};
};
};
}
);
}

118
nixos/tests/acme/dns01.nix Normal file
View file

@ -0,0 +1,118 @@
{
config,
lib,
pkgs,
...
}:
let
domain = "example.test";
dnsServerIP = nodes: nodes.dnsserver.networking.primaryIPAddress;
dnsScript = pkgs.writeShellScript "dns-hook.sh" ''
set -euo pipefail
echo '[INFO]' "[$2]" 'dns-hook.sh' $*
if [ "$1" = "present" ]; then
${pkgs.curl}/bin/curl --data @- http://dnsserver.test:8055/set-txt << EOF
{"host": "$2", "value": "$3"}
EOF
else
${pkgs.curl}/bin/curl --data @- http://dnsserver.test:8055/clear-txt << EOF
{"host": "$2"}
EOF
fi
'';
in
{
name = "dns01";
meta = {
maintainers = lib.teams.acme.members;
# Hard timeout in seconds. Average run time is about 60 seconds.
timeout = 180;
};
nodes = {
# The fake ACME server which will respond to client requests
acme =
{ nodes, ... }:
{
imports = [ ../common/acme/server ];
networking.nameservers = lib.mkForce [ (dnsServerIP nodes) ];
};
# A fake DNS server which can be configured with records as desired
# Used to test DNS-01 challenge
dnsserver =
{ nodes, ... }:
{
networking = {
firewall.allowedTCPPorts = [
8055
53
];
firewall.allowedUDPPorts = [ 53 ];
# nixos/lib/testing/network.nix will provide name resolution via /etc/hosts
# for all nodes based on their host names and domain
hostName = "dnsserver";
domain = "test";
};
systemd.services.pebble-challtestsrv = {
enable = true;
description = "Pebble ACME challenge test server";
wantedBy = [ "network.target" ];
serviceConfig = {
ExecStart = "${pkgs.pebble}/bin/pebble-challtestsrv -dns01 ':53' -defaultIPv6 '' -defaultIPv4 '${nodes.client.networking.primaryIPAddress}'";
# Required to bind on privileged ports.
AmbientCapabilities = [ "CAP_NET_BIND_SERVICE" ];
};
};
};
client =
{ nodes, ... }:
{
imports = [ ../common/acme/client ];
networking.domain = domain;
networking.nameservers = lib.mkForce [ (dnsServerIP nodes) ];
# OpenSSL will be used for more thorough certificate validation
environment.systemPackages = [ pkgs.openssl ];
security.acme.certs."${domain}" = {
domain = "*.${domain}";
dnsProvider = "exec";
dnsPropagationCheck = false;
environmentFile = pkgs.writeText "wildcard.env" ''
EXEC_PATH=${dnsScript}
EXEC_POLLING_INTERVAL=1
EXEC_PROPAGATION_TIMEOUT=1
EXEC_SEQUENCE_INTERVAL=1
'';
};
};
};
testScript = ''
${(import ./utils.nix).pythonUtils}
cert = "${domain}"
dnsserver.start()
acme.start()
wait_for_running(dnsserver)
dnsserver.wait_for_open_port(53)
wait_for_running(acme)
acme.wait_for_open_port(443)
with subtest("Boot and acquire a new cert"):
client.start()
wait_for_running(client)
check_issuer(client, cert, "pebble")
check_domain(client, cert, cert, fail=True)
check_domain(client, cert, f"toodeep.nesting.{cert}", fail=True)
check_domain(client, cert, f"whatever.{cert}")
'';
}

215
nixos/tests/acme/http01-builtin.nix Normal file
View file

@ -0,0 +1,215 @@
{
config,
lib,
pkgs,
...
}:
let
domain = "example.test";
in
{
name = "http01-builtin";
meta = {
maintainers = lib.teams.acme.members;
# Hard timeout in seconds. Average run time is about 90 seconds.
timeout = 300;
};
nodes = {
# The fake ACME server which will respond to client requests
acme =
{ nodes, ... }:
{
imports = [ ../common/acme/server ];
};
builtin =
{ nodes, config, ... }:
{
imports = [ ../common/acme/client ];
networking.domain = domain;
networking.firewall.allowedTCPPorts = [ 80 ];
# OpenSSL will be used for more thorough certificate validation
environment.systemPackages = [ pkgs.openssl ];
security.acme.certs."${config.networking.fqdn}" = {
listenHTTP = ":80";
};
specialisation = {
renew.configuration = {
# Pebble provides 5 year long certs,
# needs to be higher than that to test renewal
security.acme.certs."${config.networking.fqdn}".validMinDays = 9999;
};
accountchange.configuration = {
security.acme.certs."${config.networking.fqdn}".email = "admin@example.test";
};
keytype.configuration = {
security.acme.certs."${config.networking.fqdn}".keyType = "ec384";
};
# Perform http-01 test again, but using the pre-24.05 account hashing
# (see https://github.com/NixOS/nixpkgs/pull/317257)
# The hash is deterministic in this case - only based on keyType and email.
# Note: This test is making the assumption that the acme module will create
# the account directory regardless of internet connectivity or server reachability.
legacy_account_hash.configuration = {
security.acme.defaults.server = lib.mkForce null;
};
ocsp_stapling.configuration = {
security.acme.certs."${config.networking.fqdn}".ocspMustStaple = true;
};
preservation.configuration = { };
add_cert_and_domain.configuration = {
security.acme.certs = {
"${config.networking.fqdn}" = {
extraDomainNames = [
"builtin-alt.${domain}"
];
};
# We can assume that if renewal succeeds then the account creation leader
# logic is working, since only one service could bind to port 80 at the same time.
"builtin-2.${domain}".listenHTTP = ":80";
};
# To make sure it's the account creation leader that is doing the work.
security.acme.maxConcurrentRenewals = 10;
};
concurrency.configuration = {
# As above, relying on port binding behaviour to assert that concurrency limit
# prevents > 1 service running at a time.
security.acme.maxConcurrentRenewals = 1;
security.acme.certs = {
"${config.networking.fqdn}" = {
extraDomainNames = [
"builtin-alt.${domain}"
];
};
"builtin-2.${domain}" = {
extraDomainNames = [ "builtin-2-alt.${domain}" ];
listenHTTP = ":80";
};
"builtin-3.${domain}".listenHTTP = ":80";
};
};
};
};
};
testScript =
{ nodes, ... }:
let
certName = nodes.builtin.networking.fqdn;
caDomain = nodes.acme.test-support.acme.caDomain;
in
''
${(import ./utils.nix).pythonUtils}
domain = "${domain}"
cert = "${certName}"
cert2 = "builtin-2." + domain
cert3 = "builtin-3." + domain
legacy_account_dir = "/var/lib/acme/.lego/accounts/1ccf607d9aa280e9af00"
acme.start()
wait_for_running(acme)
acme.wait_for_open_port(443)
with subtest("Boot and acquire a new cert"):
builtin.start()
wait_for_running(builtin)
check_issuer(builtin, cert, "pebble")
check_domain(builtin, cert, cert)
with subtest("Validate permissions"):
check_permissions(builtin, cert, "acme")
with subtest("Check renewal behaviour"):
# First, test no-op behaviour
hash = builtin.succeed(f"sha256sum /var/lib/acme/{cert}/cert.pem")
# old_hash will be used in the preservation tests later
old_hash = hash
builtin.succeed(f"systemctl start acme-{cert}.service")
hash_after = builtin.succeed(f"sha256sum /var/lib/acme/{cert}/cert.pem")
assert hash == hash_after, "Certificate was unexpectedly changed"
switch_to(builtin, "renew")
check_issuer(builtin, cert, "pebble")
hash_after = builtin.succeed(f"sha256sum /var/lib/acme/{cert}/cert.pem | tee /dev/stderr")
assert hash != hash_after, "Certificate was not renewed"
with subtest("Handles email change correctly"):
hash = builtin.succeed(f"sha256sum /var/lib/acme/{cert}/cert.pem")
switch_to(builtin, "accountchange")
check_issuer(builtin, cert, "pebble")
# Check that there are now 2 account directories
builtin.succeed("test $(ls -1 /var/lib/acme/.lego/accounts | tee /dev/stderr | wc -l) -eq 2")
hash_after = builtin.succeed(f"sha256sum /var/lib/acme/{cert}/cert.pem")
# Has to do a full run to register account, which creates new certs.
assert hash != hash_after, "Certificate was not renewed"
# Remove the new account directory
builtin.succeed(
"cd /var/lib/acme/.lego/accounts"
" && ls -1 --sort=time | tee /dev/stderr | head -1 | xargs rm -rf"
)
# old_hash will be used in the preservation tests later
old_hash = hash_after
with subtest("Correctly implements OCSP stapling"):
check_stapling(builtin, cert, "${caDomain}", fail=True)
switch_to(builtin, "ocsp_stapling")
check_stapling(builtin, cert, "${caDomain}")
with subtest("Handles keyType change correctly"):
check_key_bits(builtin, cert, 256)
switch_to(builtin, "keytype")
check_key_bits(builtin, cert, 384)
# keyType is part of the accountHash, thus a new account will be created
builtin.succeed("test $(ls -1 /var/lib/acme/.lego/accounts | tee /dev/stderr | wc -l) -eq 2")
with subtest("Reuses generated, valid certs from previous configurations"):
# Right now, the hash should not match due to the previous test
hash = builtin.succeed(f"sha256sum /var/lib/acme/{cert}/cert.pem | tee /dev/stderr")
assert hash != old_hash, "Expected certificate to differ"
switch_to(builtin, "preservation")
hash = builtin.succeed(f"sha256sum /var/lib/acme/{cert}/cert.pem | tee /dev/stderr")
assert hash == old_hash, "Expected certificate to match from older configuration"
with subtest("Add a new cert, extend existing cert domains"):
check_domain(builtin, cert, f"builtin-alt.{domain}", fail=True)
switch_to(builtin, "add_cert_and_domain")
check_issuer(builtin, cert, "pebble")
check_domain(builtin, cert, f"builtin-alt.{domain}")
check_issuer(builtin, cert2, "pebble")
check_domain(builtin, cert2, cert2)
# There should not be a new account folder created
builtin.succeed("test $(ls -1 /var/lib/acme/.lego/accounts | tee /dev/stderr | wc -l) -eq 2")
with subtest("Check account hashing compatibility with pre-24.05 settings"):
switch_to(builtin, "legacy_account_hash", fail=True)
builtin.succeed(f"stat {legacy_account_dir} > /dev/stderr && rm -rf {legacy_account_dir}")
with subtest("Ensure Concurrency limits work"):
switch_to(builtin, "concurrency")
check_issuer(builtin, cert3, "pebble")
check_domain(builtin, cert3, cert3)
with subtest("Generate self-signed certs"):
check_issuer(builtin, cert, "pebble")
builtin.succeed(f"systemctl clean acme-{cert}.service --what=state")
builtin.succeed(f"systemctl start acme-selfsigned-{cert}.service")
check_issuer(builtin, cert, "minica")
check_domain(builtin, cert, cert)
with subtest("Validate permissions (self-signed)"):
check_permissions(builtin, cert, "acme")
'';
}

166
nixos/tests/acme/python-utils.py Normal file
View file

@ -0,0 +1,166 @@
#!/usr/bin/env python3
import time
TOTAL_RETRIES = 20
def run(node, cmd, fail=False):
if fail:
return node.fail(cmd)
else:
return node.succeed(cmd)
# Waits for the system to finish booting or switching configuration
def wait_for_running(node):
node.succeed("systemctl is-system-running --wait")
# On first switch, this will create a symlink to the current system so that we can
# quickly switch between derivations
def switch_to(node, name, fail=False) -> None:
root_specs = "/tmp/specialisation"
node.execute(
f"test -e {root_specs}"
f" || ln -s $(readlink /run/current-system)/specialisation {root_specs}"
)
switcher_path = (
f"/run/current-system/specialisation/{name}/bin/switch-to-configuration"
)
rc, _ = node.execute(f"test -e '{switcher_path}'")
if rc > 0:
switcher_path = f"/tmp/specialisation/{name}/bin/switch-to-configuration"
cmd = f"{switcher_path} test"
run(node, cmd, fail=fail)
if not fail:
wait_for_running(node)
# Ensures the issuer of our cert matches the chain
# and matches the issuer we expect it to be.
# It's a good validation to ensure the cert.pem and fullchain.pem
# are not still selfsigned after verification
def check_issuer(node, cert_name, issuer) -> None:
for fname in ("cert.pem", "fullchain.pem"):
actual_issuer = node.succeed(
f"openssl x509 -noout -issuer -in /var/lib/acme/{cert_name}/{fname}"
).partition("=")[2]
assert (
issuer.lower() in actual_issuer.lower()
), f"{fname} issuer mismatch. Expected {issuer} got {actual_issuer}"
# Ensures the provided domain matches with the given cert
def check_domain(node, cert_name, domain, fail=False) -> None:
cmd = f"openssl x509 -noout -checkhost '{domain}' -in /var/lib/acme/{cert_name}/cert.pem"
run(node, cmd, fail=fail)
# Ensures the required values for OCSP stapling are present
# Pebble doesn't provide a full OCSP responder, so just checks the URL
def check_stapling(node, cert_name, ca_domain, fail=False):
rc, _ = node.execute(
f"openssl x509 -noout -ocsp_uri -in /var/lib/acme/{cert_name}/cert.pem"
f" | grep -i 'http://{ca_domain}:4002' 2>&1",
)
assert rc == 0 or fail, "Failed to find OCSP URI in issued certificate"
run(
node,
f"openssl x509 -noout -ext tlsfeature -in /var/lib/acme/{cert_name}/cert.pem"
f" | grep -iv 'no extensions' 2>&1",
fail=fail,
)
# Checks the keyType by validating the number of bits
def check_key_bits(node, cert_name, bits, fail=False):
run(
node,
f"openssl x509 -noout -text -in /var/lib/acme/{cert_name}/cert.pem"
f" | grep -i Public-Key | grep {bits} | tee /dev/stderr",
fail=fail,
)
# Ensure cert comes before chain in fullchain.pem
def check_fullchain(node, cert_name):
cert_file = f"/var/lib/acme/{cert_name}/fullchain.pem"
num_certs = node.succeed(f"grep -o 'END CERTIFICATE' {cert_file}")
assert len(num_certs.strip().split("\n")) > 1, "Insufficient certs in fullchain.pem"
first_cert_data = node.succeed(
f"grep -m1 -B50 'END CERTIFICATE' {cert_file}"
" | openssl x509 -noout -text"
)
for line in first_cert_data.lower().split("\n"):
if "dns:" in line:
print(f"First DNSName in fullchain.pem: {line}")
assert cert_name.lower() in line, f"{cert_name} not found in {line}"
return
assert False
# Checks the permissions in the cert directories are as expected
def check_permissions(node, cert_name, group):
stat = "stat -L -c '%a %U %G' "
node.succeed(
f"test $({stat} /var/lib/acme/{cert_name}/*.pem"
f" | tee /dev/stderr | grep -v '640 acme {group}' | wc -l) -eq 0"
)
node.succeed(
f"test $({stat} /var/lib/acme/.lego/{cert_name}/*/{cert_name}*"
f" | tee /dev/stderr | grep -v '600 acme {group}' | wc -l) -eq 0"
)
node.succeed(
f"test $({stat} /var/lib/acme/{cert_name}"
f" | tee /dev/stderr | grep -v '750 acme {group}' | wc -l) -eq 0"
)
node.succeed(
f"test $(find /var/lib/acme/.lego/accounts -type f -exec {stat} {{}} \\;"
f" | tee /dev/stderr | grep -v '600 acme {group}' | wc -l) -eq 0"
)
# BackoffTracker provides a robust system for handling test retries
class BackoffTracker:
delay = 1
increment = 1
def handle_fail(self, retries, message) -> int:
assert retries < TOTAL_RETRIES, message
print(f"Retrying in {self.delay}s, {retries + 1}/{TOTAL_RETRIES}")
time.sleep(self.delay)
# Only increment after the first try
if retries == 0:
self.delay += self.increment
self.increment *= 2
return retries + 1
def protect(self, func):
def wrapper(*args, retries: int = 0, **kwargs):
try:
return func(*args, **kwargs)
except Exception as err:
retries = self.handle_fail(retries, err.args)
return wrapper(*args, retries=retries, **kwargs)
return wrapper
backoff = BackoffTracker()
@backoff.protect
def download_ca_certs(node, ca_domain):
node.succeed(f"curl https://{ca_domain}:15000/roots/0 > /tmp/ca.crt")
node.succeed(f"curl https://{ca_domain}:15000/intermediate-keys/0 >> /tmp/ca.crt")
@backoff.protect
def check_connection(node, domain, fail=False, minica=False):
cafile = "/tmp/ca.crt"
if minica:
cafile = "/var/lib/acme/.minica/cert.pem"
run(node,
f"openssl s_client -brief -CAfile {cafile}"
f" -verify 2 -verify_return_error -verify_hostname {domain}"
f" -servername {domain} -connect {domain}:443 < /dev/null",
fail=fail,
)

4
nixos/tests/acme/utils.nix Normal file
View file

@ -0,0 +1,4 @@
{
# Helper functions for python
pythonUtils = builtins.readFile ./python-utils.py;
}

185
nixos/tests/acme/webserver.nix Normal file
View file

@ -0,0 +1,185 @@
{
serverName,
group,
baseModule,
domain ? "example.test",
}:
{
config,
lib,
pkgs,
...
}:
{
name = serverName;
meta = {
maintainers = lib.teams.acme.members;
# Hard timeout in seconds. Average run time is about 100 seconds.
timeout = 300;
};
nodes = {
# The fake ACME server which will respond to client requests
acme =
{ nodes, ... }:
{
imports = [ ../common/acme/server ];
};
webserver =
{ nodes, ... }:
{
imports = [
../common/acme/client
baseModule
];
networking.domain = domain;
networking.firewall.allowedTCPPorts = [
80
443
];
# Resolve the vhosts the easy way
networking.hosts."127.0.0.1" = [
"proxied.${domain}"
"certchange.${domain}"
"zeroconf.${domain}"
"zeroconf2.${domain}"
"nullroot.${domain}"
];
# OpenSSL will be used for more thorough certificate validation
environment.systemPackages = [ pkgs.openssl ];
# Used to determine if service reload was triggered.
# This does not provide a guarantee that the webserver is finished reloading,
# to handle that there is retry logic wrapping any connectivity checks.
systemd.targets."renew-triggered" = {
wantedBy = [ "${serverName}-config-reload.service" ];
after = [ "${serverName}-config-reload.service" ];
};
security.acme.certs."proxied.${domain}" = {
listenHTTP = ":8080";
group = group;
};
specialisation = {
# Test that the web server is correctly reloaded when the cert changes
certchange.configuration = {
security.acme.certs."proxied.${domain}".extraDomainNames = [
"certchange.${domain}"
];
};
# A useful transitional step before other tests, and tests behaviour
# of removing an extra domain from a cert.
certundo.configuration = { };
# Tests these features:
# - enableACME behaves as expected
# - serverAliases are appended to extraDomainNames
# - Correct routing to the specific virtualHost for a cert
# Inherits previous test config
zeroconf.configuration = {
services.${serverName}.virtualHosts."zeroconf.${domain}" = {
addSSL = true;
enableACME = true;
serverAliases = [ "zeroconf2.${domain}" ];
};
};
# Test that serverAliases are correctly removed which triggers
# cert regeneration and service reload.
rmalias.configuration = {
services.${serverName}.virtualHosts."zeroconf.${domain}" = {
addSSL = true;
enableACME = true;
};
};
# Test that "acmeRoot = null" still results in
# valid cert generation by inheriting defaults.
nullroot.configuration = {
security.acme.defaults.listenHTTP = ":8080";
services.${serverName}.virtualHosts."nullroot.${domain}" = {
onlySSL = true;
enableACME = true;
acmeRoot = null;
};
};
};
};
};
testScript =
{ nodes, ... }:
''
${(import ./utils.nix).pythonUtils}
domain = "${domain}"
ca_domain = "${nodes.acme.test-support.acme.caDomain}"
fqdn = f"proxied.{domain}"
acme.start()
wait_for_running(acme)
acme.wait_for_open_port(443)
with subtest("Acquire a cert through a proxied lego"):
webserver.start()
webserver.succeed("systemctl is-system-running --wait")
wait_for_running(webserver)
download_ca_certs(webserver, ca_domain)
check_connection(webserver, fqdn)
with subtest("Can run on selfsigned certificates"):
# Switch to selfsigned first
webserver.succeed(f"systemctl clean acme-{fqdn}.service --what=state")
webserver.succeed(f"systemctl start acme-selfsigned-{fqdn}.service")
check_issuer(webserver, fqdn, "minica")
webserver.succeed("systemctl restart ${serverName}-config-reload.service")
# Check that the web server has picked up the selfsigned cert
check_connection(webserver, fqdn, minica=True)
webserver.succeed("systemctl stop renew-triggered.target")
webserver.succeed(f"systemctl start acme-{fqdn}.service")
webserver.wait_for_unit("renew-triggered.target")
check_issuer(webserver, fqdn, "pebble")
check_connection(webserver, fqdn)
with subtest("security.acme changes reflect on web server part 1"):
check_connection(webserver, f"certchange.{domain}", fail=True)
switch_to(webserver, "certchange")
webserver.wait_for_unit("renew-triggered.target")
check_connection(webserver, f"certchange.{domain}")
check_connection(webserver, fqdn)
with subtest("security.acme changes reflect on web server part 2"):
check_connection(webserver, f"certchange.{domain}")
switch_to(webserver, "certundo")
webserver.wait_for_unit("renew-triggered.target")
check_connection(webserver, f"certchange.{domain}", fail=True)
check_connection(webserver, fqdn)
with subtest("Zero configuration SSL certificates for a vhost"):
check_connection(webserver, f"zeroconf.{domain}", fail=True)
switch_to(webserver, "zeroconf")
webserver.wait_for_unit("renew-triggered.target")
check_connection(webserver, f"zeroconf.{domain}")
check_connection(webserver, f"zeroconf2.{domain}")
check_connection(webserver, fqdn)
with subtest("Removing an alias from a vhost"):
check_connection(webserver, f"zeroconf2.{domain}")
switch_to(webserver, "rmalias")
webserver.wait_for_unit("renew-triggered.target")
check_connection(webserver, f"zeroconf2.{domain}", fail=True)
check_connection(webserver, f"zeroconf.{domain}")
check_connection(webserver, fqdn)
with subtest("Create cert using inherited default validation mechanism"):
check_connection(webserver, f"nullroot.{domain}", fail=True)
switch_to(webserver, "nullroot")
webserver.wait_for_unit("renew-triggered.target")
check_connection(webserver, f"nullroot.{domain}")
'';
}

2
nixos/tests/all-tests.nix
View file

@ -105,7 +105,7 @@ in {
_3proxy = runTest ./3proxy.nix;
aaaaxy = runTest ./aaaaxy.nix;
acme = runTest ./acme.nix;
acme = import ./acme/default.nix { inherit runTest; };
acme-dns = handleTest ./acme-dns.nix {};
actual = handleTest ./actual.nix {};
adguardhome = runTest ./adguardhome.nix;

111
nixos/tests/common/acme/server/default.nix
View file

@ -6,53 +6,15 @@
# the test certificate into security.pki.certificateFiles or into package
# overlays.
#
# Another value that's needed if you don't use a custom resolver (see below for
# notes on that) is to add the acme node as a nameserver to every node
# that needs to acquire certificates using ACME, because otherwise the API host
# for acme.test can't be resolved.
#
# A configuration example of a full node setup using this would be this:
#
# {
# acme = import ./common/acme/server;
#
# example = { nodes, ... }: {
# networking.nameservers = [
# nodes.acme.networking.primaryIPAddress
# ];
# security.pki.certificateFiles = [
# nodes.acme.test-support.acme.caCert
# ];
# };
# }
#
# By default, this module runs a local resolver, generated using resolver.nix
# from the parent directory to automatically discover all zones in the network.
#
# If you do not want this and want to use your own resolver, you can just
# override networking.nameservers like this:
#
# {
# acme = { nodes, lib, ... }: {
# imports = [ ./common/acme/server ];
# networking.nameservers = lib.mkForce [
# nodes.myresolver.networking.primaryIPAddress
# ];
# };
#
# myresolver = ...;
# }
#
# Keep in mind, that currently only _one_ resolver is supported, if you have
# more than one resolver in networking.nameservers only the first one will be
# used.
#
# Also make sure that whenever you use a resolver from a different test node
# that it has to be started _before_ the ACME service.
# The hosts file of this node will be populated with a mapping of certificate
# domains (including extraDomainNames) to their parent nodes in the test suite.
# This negates the need for a DNS server for most testing. You can still specify
# a custom nameserver/resolver if necessary for other reasons.
{
config,
pkgs,
lib,
nodes ? { },
...
}:
let
@ -75,8 +37,6 @@ let
in
{
imports = [ ../../resolver.nix ];
options.test-support.acme = {
caDomain = lib.mkOption {
type = lib.types.str;
@ -100,29 +60,49 @@ in
};
config = {
test-support = {
resolver.enable =
networking = {
firewall.allowedTCPPorts = [
80
443
15000
4002
];
# Match the caDomain - nixos/lib/testing/network.nix will then add a record for us to
# all nodes in /etc/hosts
hostName = "acme";
domain = "test";
# Extend /etc/hosts to resolve all configured certificates to their hosts.
# This way, no DNS server will be needed to validate HTTP-01 certs.
hosts = lib.attrsets.concatMapAttrs (
_: node:
let
isLocalResolver = config.networking.nameservers == [ "127.0.0.1" ];
inherit (node.networking) primaryIPAddress primaryIPv6Address;
ips = builtins.filter (ip: ip != "") [
primaryIPAddress
primaryIPv6Address
];
names = lib.lists.unique (
lib.lists.flatten (
lib.lists.concatMap
(
cfg:
lib.attrsets.mapAttrsToList (
domain: cfg:
builtins.map (builtins.replaceStrings [ "*." ] [ "" ]) ([ domain ] ++ cfg.extraDomainNames)
) cfg.configuration.security.acme.certs
)
# A specialisation's config is nested under its configuration attribute.
# For ease of use, nest the root node's configuration simiarly.
([ { configuration = node; } ] ++ (builtins.attrValues node.specialisation))
)
);
in
lib.mkOverride 900 isLocalResolver;
builtins.listToAttrs (builtins.map (ip: lib.attrsets.nameValuePair ip names) ips)
) nodes;
};
# This has priority 140, because modules/testing/test-instrumentation.nix
# already overrides this with priority 150.
networking.nameservers = lib.mkOverride 140 [ "127.0.0.1" ];
networking.firewall.allowedTCPPorts = [
80
443
15000
4002
];
networking.extraHosts = ''
127.0.0.1 ${domain}
${config.networking.primaryIPAddress} ${domain}
'';
systemd.services = {
pebble = {
enable = true;
@ -130,8 +110,9 @@ in
wantedBy = [ "network.target" ];
environment = {
# We're not testing lego, we're just testing our configuration.
# No need to sleep.
# No need to sleep or randomly fail nonces.
PEBBLE_VA_NOSLEEP = "1";
PEBBLE_WFE_NONCEREJECT = "0";
};
serviceConfig = {

184
nixos/tests/common/resolver.nix
View file

@ -1,184 +0,0 @@
# This module automatically discovers zones in BIND and NSD NixOS
# configurations and creates zones for all definitions of networking.extraHosts
# (except those that point to 127.0.0.1 or ::1) within the current test network
# and delegates these zones using a fake root zone served by a BIND recursive
# name server.
{
config,
nodes,
pkgs,
lib,
...
}:
{
options.test-support.resolver.enable = lib.mkOption {
type = lib.types.bool;
default = true;
internal = true;
description = ''
Whether to enable the resolver that automatically discovers zone in the
test network.
This option is `true` by default, because the module
defining this option needs to be explicitly imported.
The reason this option exists is for the
{file}`nixos/tests/common/acme/server` module, which
needs that option to disable the resolver once the user has set its own
resolver.
'';
};
config = lib.mkIf config.test-support.resolver.enable {
networking.firewall.enable = false;
services.bind.enable = true;
services.bind.cacheNetworks = lib.mkForce [ "any" ];
services.bind.forwarders = lib.mkForce [ ];
services.bind.zones = lib.singleton {
name = ".";
master = true;
file =
let
addDot = zone: zone + lib.optionalString (!lib.hasSuffix "." zone) ".";
mkNsdZoneNames = zones: map addDot (lib.attrNames zones);
mkBindZoneNames = zones: map addDot (lib.attrNames zones);
getZones = cfg: mkNsdZoneNames cfg.services.nsd.zones ++ mkBindZoneNames cfg.services.bind.zones;
getZonesForNode = attrs: {
ip = attrs.config.networking.primaryIPAddress;
zones = lib.filter (zone: zone != ".") (getZones attrs.config);
};
zoneInfo = lib.mapAttrsToList (lib.const getZonesForNode) nodes;
# A and AAAA resource records for all the definitions of
# networking.extraHosts except those for 127.0.0.1 or ::1.
#
# The result is an attribute set with keys being the host name and the
# values are either { ipv4 = ADDR; } or { ipv6 = ADDR; } where ADDR is
# the IP address for the corresponding key.
recordsFromExtraHosts =
let
getHostsForNode = lib.const (n: n.config.networking.extraHosts);
allHostsList = lib.mapAttrsToList getHostsForNode nodes;
allHosts = lib.concatStringsSep "\n" allHostsList;
reIp = "[a-fA-F0-9.:]+";
reHost = "[a-zA-Z0-9.-]+";
matchAliases =
str:
let
matched = builtins.match "[ \t]+(${reHost})(.*)" str;
continue = lib.singleton (lib.head matched) ++ matchAliases (lib.last matched);
in
lib.optional (matched != null) continue;
matchLine =
str:
let
result = builtins.match "[ \t]*(${reIp})[ \t]+(${reHost})(.*)" str;
in
if result == null then
null
else
{
ipAddr = lib.head result;
hosts = lib.singleton (lib.elemAt result 1) ++ matchAliases (lib.last result);
};
skipLine =
str:
let
rest = builtins.match "[^\n]*\n(.*)" str;
in
if rest == null then "" else lib.head rest;
getEntries =
str: acc:
let
result = matchLine str;
next = getEntries (skipLine str);
newEntry = acc ++ lib.singleton result;
continue = if result == null then next acc else next newEntry;
in
if str == "" then acc else continue;
isIPv6 = str: builtins.match ".*:.*" str != null;
loopbackIps = [
"127.0.0.1"
"::1"
];
filterLoopback = lib.filter (e: !lib.elem e.ipAddr loopbackIps);
allEntries = lib.concatMap (
entry:
map (host: {
inherit host;
${if isIPv6 entry.ipAddr then "ipv6" else "ipv4"} = entry.ipAddr;
}) entry.hosts
) (filterLoopback (getEntries (allHosts + "\n") [ ]));
mkRecords =
entry:
let
records =
lib.optional (entry ? ipv6) "AAAA ${entry.ipv6}"
++ lib.optional (entry ? ipv4) "A ${entry.ipv4}";
mkRecord = typeAndData: "${entry.host}. IN ${typeAndData}";
in
lib.concatMapStringsSep "\n" mkRecord records;
in
lib.concatMapStringsSep "\n" mkRecords allEntries;
# All of the zones that are subdomains of existing zones.
# For example if there is only "example.com" the following zones would
# be 'subZones':
#
# * foo.example.com.
# * bar.example.com.
#
# While the following would *not* be 'subZones':
#
# * example.com.
# * com.
#
subZones =
let
allZones = lib.concatMap (zi: zi.zones) zoneInfo;
isSubZoneOf = z1: z2: lib.hasSuffix z2 z1 && z1 != z2;
in
lib.filter (z: lib.any (isSubZoneOf z) allZones) allZones;
# All the zones without 'subZones'.
filteredZoneInfo = map (
zi:
zi
// {
zones = lib.filter (x: !lib.elem x subZones) zi.zones;
}
) zoneInfo;
in
pkgs.writeText "fake-root.zone" ''
$TTL 3600
. IN SOA ns.fakedns. admin.fakedns. ( 1 3h 1h 1w 1d )
ns.fakedns. IN A ${config.networking.primaryIPAddress}
. IN NS ns.fakedns.
${lib.concatImapStrings (
num:
{ ip, zones }:
''
ns${toString num}.fakedns. IN A ${ip}
${lib.concatMapStrings (zone: ''
${zone} IN NS ns${toString num}.fakedns.
'') zones}
''
) (lib.filter (zi: zi.zones != [ ]) filteredZoneInfo)}
${recordsFromExtraHosts}
'';
};
};
}

1
pkgs/by-name/ca/caddy/package.nix
View file

@ -77,6 +77,7 @@ buildGoModule {
command = "${caddy}/bin/caddy version";
package = caddy;
};
acme-integration = nixosTests.acme.caddy;
};
withPlugins = callPackage ./plugins.nix { inherit caddy; };
};

5
pkgs/by-name/le/lego/package.nix
View file

@ -29,5 +29,8 @@ buildGoModule rec {
mainProgram = "lego";
};
passthru.tests.lego = nixosTests.acme;
passthru.tests = {
lego-http = nixosTests.acme.http01-builtin;
lego-dns = nixosTests.acme.dns01-builtin;
};
}

3
pkgs/by-name/pe/pebble/package.nix
View file

@ -25,7 +25,8 @@ buildGoModule rec {
];
passthru.tests = {
smoke-test = nixosTests.acme;
smoke-test-http = nixosTests.acme.http01-builtin;
smoke-test-dns = nixosTests.acme.dns01;
};
meta = {

2
pkgs/servers/http/apache-httpd/2.4.nix
View file

@ -99,7 +99,7 @@ stdenv.mkDerivation rec {
passthru = {
inherit apr aprutil sslSupport proxySupport ldapSupport luaSupport lua5;
tests = {
acme-integration = nixosTests.acme;
acme-integration = nixosTests.acme.httpd;
proxy = nixosTests.proxy;
php = nixosTests.php.httpd;
cross = runCommand "apacheHttpd-test-cross" { } ''

2
pkgs/servers/http/nginx/generic.nix
View file

@ -300,7 +300,7 @@ stdenv.mkDerivation {
nginx-unix-socket
;
variants = lib.recurseIntoAttrs nixosTests.nginx-variants;
acme-integration = nixosTests.acme;
acme-integration = nixosTests.acme.nginx;
} // passthru.tests;
};