✨ Adds MariaDB image with direct LDAP PAM authentication
- Introduced a new Containerfile to build a MariaDB image with direct LDAP PAM support. - Implemented a custom PAM module (`pam_ldap_direct.so`) built from the PADL source to avoid dependencies on `libpam-ldap` and `nslcd`. - Updated the entrypoint script to dynamically generate the PAM configuration file (`/etc/pam_ldap.conf`) based on environment variables. - Added a README.md to document the image usage, configuration, and build process. - Created a test suite to validate the image build and entrypoint behavior, ensuring proper handling of environment variables and configuration generation. - Configured health checks and ensured the stock MariaDB entrypoint and lifecycle are preserved.
This commit is contained in:
11 files changed
+519
-282
No files matched your search
@@ -0,0 +1,247 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import stat
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
MODULE = "/usr/local/lib/security/pam_ldap_direct.so"
|
||||
HELPER = "/usr/lib/mysql/plugin/auth_pam_tool_dir/auth_pam_tool"
|
||||
CONF = "/etc/pam_ldap.conf"
|
||||
CA = "/etc/ssl/certs/ca-certificates.crt"
|
||||
|
||||
|
||||
class ImageRecipeTests(unittest.TestCase):
|
||||
def test_linux_entrypoint_line_endings_are_preserved(self):
|
||||
self.assertIn("docker-entrypoint.sh text eol=lf", (ROOT / ".gitattributes").read_text())
|
||||
self.assertNotIn(b"\r", (ROOT / "docker-entrypoint.sh").read_bytes())
|
||||
|
||||
def test_source_is_pinned_and_checked_before_build(self):
|
||||
recipe = (ROOT / "Containerfile").read_text()
|
||||
self.assertIn("ARG MARIADB_VERSION=13.0.2", recipe)
|
||||
self.assertIn("656448f091cbeb9efb3ece08e6868e40b8e7b6f8", recipe)
|
||||
self.assertIn("1aa622e85621b9018025e3483956d854566a5bc0636a4e8d9d005093f02dd16b", recipe)
|
||||
self.assertLess(recipe.index("sha256sum -c -"), recipe.index("./configure"))
|
||||
self.assertIn("--with-ldap-conf-file=/etc/pam_ldap.conf", recipe)
|
||||
self.assertIn("PERL5LIB=/build/pam_ldap make", recipe)
|
||||
|
||||
def test_runtime_does_not_install_the_virtual_ldap_package(self):
|
||||
recipe = (ROOT / "Containerfile").read_text()
|
||||
self.assertNotIn("libpam-ldap", recipe)
|
||||
self.assertNotIn("nslcd", recipe)
|
||||
self.assertNotIn("mariadb-plugin-auth-pam || true", recipe)
|
||||
self.assertIn("libldap2", recipe)
|
||||
self.assertIn("util-linux", recipe)
|
||||
self.assertIn("command -v setpriv", recipe)
|
||||
self.assertIn("ldd \"$module\"", recipe)
|
||||
self.assertIn("not found", recipe)
|
||||
self.assertIn("chmod 4750", recipe)
|
||||
self.assertIn("gosu mysql test -x", recipe)
|
||||
|
||||
def test_pam_stack_selects_the_exact_direct_module(self):
|
||||
stack = (ROOT / "pam.d" / "mariadb").read_text().splitlines()
|
||||
self.assertEqual(
|
||||
[["auth", "required", MODULE], ["account", "required", MODULE]],
|
||||
[line.split() for line in stack],
|
||||
)
|
||||
|
||||
def test_stock_database_lifecycle_is_preserved(self):
|
||||
recipe = (ROOT / "Containerfile").read_text()
|
||||
entrypoint = (ROOT / "docker-entrypoint.sh").read_text()
|
||||
self.assertIn('ENTRYPOINT ["ldap-entrypoint.sh"]', recipe)
|
||||
self.assertIn('CMD ["mariadbd"]', recipe)
|
||||
self.assertIn('exec docker-entrypoint.sh "$@"', entrypoint)
|
||||
self.assertIn("healthcheck.sh --connect --innodb_initialized", recipe)
|
||||
self.assertNotIn("nslcd", entrypoint)
|
||||
self.assertIn("LC_ALL=C setpriv --dump", entrypoint)
|
||||
self.assertNotIn("/proc/self/status", entrypoint)
|
||||
|
||||
|
||||
@unittest.skipUnless(os.name == "posix" and shutil.which("bash"), "Requires a local POSIX shell; no Docker or lab access.")
|
||||
class EntrypointTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory(prefix="mariadb-pam-entrypoint-")
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.root = Path(self.temp.name)
|
||||
self.bin = self.root / "bin"
|
||||
self.bin.mkdir()
|
||||
self.log = self.root / "calls"
|
||||
self.conf = self.root / "pam_ldap.conf"
|
||||
self.module = self.root / "pam_ldap_direct.so"
|
||||
self.helper = self.root / "auth_pam_tool"
|
||||
self.ca = self.root / "ca.pem"
|
||||
self.status = self.root / "status"
|
||||
self.module.write_text("fixture-module")
|
||||
self.helper.write_text("fixture-helper")
|
||||
self.helper.chmod(0o4750)
|
||||
self.ca.write_text("fixture-ca")
|
||||
self.status.write_text("NoNewPrivs:\t0\n")
|
||||
text = (ROOT / "docker-entrypoint.sh").read_text()
|
||||
for old, new in (
|
||||
(MODULE, self.module), (HELPER, self.helper), (CONF, self.conf),
|
||||
(CA, self.ca), ("/proc/self/status", self.status),
|
||||
):
|
||||
text = text.replace(old, str(new))
|
||||
self.script = self.root / "entrypoint.sh"
|
||||
self.script.write_text(text)
|
||||
self.env = {
|
||||
**os.environ,
|
||||
"PATH": str(self.bin) + os.pathsep + os.environ["PATH"],
|
||||
"LDAPURI": "ldaps://directory.example.test",
|
||||
"LDAP_BASE": "DC=example,DC=test",
|
||||
"LDAPBINDDN": "CN=reader,DC=example,DC=test",
|
||||
"LDAPPASSWORD": "fixture-password-not-for-logs",
|
||||
"LDAP_LOGIN_ATTR": "sAMAccountName",
|
||||
"LDAP_TLS_REQCERT": "demand",
|
||||
"CERTS_GID": "12002",
|
||||
"TEST_CALLS": str(self.log),
|
||||
"TEST_HAVE_GROUP": "1",
|
||||
"TEST_UID": "0",
|
||||
"TEST_NO_NEW_PRIVS": "0",
|
||||
"TEST_SETPRIV_RC": "0",
|
||||
}
|
||||
self.stub("id", 'printf "%s\\n" "$TEST_UID"\n')
|
||||
self.stub("getent", 'if [ "$TEST_HAVE_GROUP" = 1 ]; then printf "certgroup:x:%s:\\n" "$2"; else exit 2; fi\n')
|
||||
self.stub("groupadd", 'printf "groupadd %s\\n" "$*" >> "$TEST_CALLS"\n')
|
||||
self.stub("usermod", 'printf "usermod %s\\n" "$*" >> "$TEST_CALLS"\n')
|
||||
self.stub(
|
||||
"setpriv",
|
||||
'[ "$1" = --dump ] || exit 9\n'
|
||||
'[ "$TEST_SETPRIV_RC" = 0 ] || exit "$TEST_SETPRIV_RC"\n'
|
||||
'printf "uid: 999\\nno_new_privs: %s\\n" "$TEST_NO_NEW_PRIVS"\n',
|
||||
)
|
||||
self.stub("docker-entrypoint.sh", 'printf "delegate %s\\n" "$*" >> "$TEST_CALLS"\n')
|
||||
|
||||
def stub(self, name, body):
|
||||
path = self.bin / name
|
||||
path.write_text("#!/bin/sh\nset -eu\n" + body)
|
||||
path.chmod(0o755)
|
||||
|
||||
def run_entrypoint(self, expected=0):
|
||||
result = subprocess.run(
|
||||
["bash", str(self.script), "mariadbd", "--server-id=2", "--read-only=ON"],
|
||||
env=self.env, text=True, capture_output=True, timeout=5,
|
||||
)
|
||||
self.assertEqual(expected, result.returncode, result.stdout + result.stderr)
|
||||
self.assertNotIn(self.env.get("LDAPPASSWORD", "unused-secret"), result.stdout + result.stderr)
|
||||
return result
|
||||
|
||||
def test_ldaps_writes_only_the_direct_module_configuration(self):
|
||||
self.run_entrypoint()
|
||||
text = self.conf.read_text()
|
||||
self.assertIn("uri ldaps://directory.example.test\n", text)
|
||||
self.assertIn("ssl on\n", text)
|
||||
self.assertIn("tls_checkpeer yes\n", text)
|
||||
self.assertIn(f"tls_cacertfile {self.ca}\n", text)
|
||||
self.assertIn("pam_login_attribute sAMAccountName\n", text)
|
||||
self.assertIn("bindpw fixture-password-not-for-logs\n", text)
|
||||
self.assertNotIn("tls_reqcert", text)
|
||||
self.assertEqual(0o600, stat.S_IMODE(self.conf.stat().st_mode))
|
||||
self.assertIn("usermod -aG certgroup mysql", self.log.read_text())
|
||||
self.assertIn("delegate mariadbd --server-id=2 --read-only=ON", self.log.read_text())
|
||||
|
||||
def test_ldap_uses_starttls_not_plain_bind(self):
|
||||
self.env["LDAPURI"] = "ldap://directory.example.test:389"
|
||||
self.run_entrypoint()
|
||||
self.assertIn("ssl start_tls\n", self.conf.read_text())
|
||||
self.assertIn("tls_checkpeer yes\n", self.conf.read_text())
|
||||
|
||||
def test_multiple_same_transport_uris_are_supported(self):
|
||||
self.env["LDAPURI"] = "ldaps://one.example.test ldaps://two.example.test"
|
||||
self.run_entrypoint()
|
||||
self.assertIn("uri " + self.env["LDAPURI"], self.conf.read_text())
|
||||
|
||||
def test_mixed_transports_are_rejected(self):
|
||||
self.env["LDAPURI"] = "ldaps://one.example.test ldap://two.example.test"
|
||||
result = self.run_entrypoint(1)
|
||||
self.assertIn("cannot mix", result.stderr)
|
||||
self.assertFalse(self.log.exists())
|
||||
|
||||
def test_weak_tls_verification_is_rejected(self):
|
||||
for value in ("allow", "try", "never"):
|
||||
with self.subTest(value=value):
|
||||
self.env["LDAP_TLS_REQCERT"] = value
|
||||
self.run_entrypoint(1)
|
||||
self.assertFalse(self.log.exists())
|
||||
|
||||
def test_hard_is_an_alias_for_verified_tls(self):
|
||||
self.env["LDAP_TLS_REQCERT"] = "hard"
|
||||
self.run_entrypoint()
|
||||
self.assertIn("tls_checkpeer yes", self.conf.read_text())
|
||||
|
||||
def test_missing_secret_is_not_defaulted(self):
|
||||
del self.env["LDAPPASSWORD"]
|
||||
result = self.run_entrypoint(1)
|
||||
self.assertIn("LDAPPASSWORD is required", result.stderr)
|
||||
self.assertFalse(self.conf.exists())
|
||||
|
||||
def test_newline_injection_fails_without_logging_the_secret(self):
|
||||
self.env["LDAPPASSWORD"] = "fixture-password-not-for-logs\ntls_checkpeer no"
|
||||
result = self.run_entrypoint(1)
|
||||
self.assertIn("must be a single line", result.stderr)
|
||||
self.assertNotIn("fixture-password-not-for-logs", result.stderr)
|
||||
self.assertFalse(self.conf.exists())
|
||||
|
||||
def test_missing_or_non_setuid_helper_is_reported_before_start(self):
|
||||
self.helper.chmod(0o0750)
|
||||
result = self.run_entrypoint(1)
|
||||
self.assertIn("setuid PAM helper", result.stderr)
|
||||
self.assertFalse(self.log.exists())
|
||||
|
||||
def test_missing_module_is_reported_before_start(self):
|
||||
self.module.unlink()
|
||||
result = self.run_entrypoint(1)
|
||||
self.assertIn("direct LDAP PAM module is missing", result.stderr)
|
||||
self.assertFalse(self.conf.exists())
|
||||
|
||||
def test_missing_trust_store_fails_closed(self):
|
||||
self.ca.unlink()
|
||||
result = self.run_entrypoint(1)
|
||||
self.assertIn("system CA bundle", result.stderr)
|
||||
self.assertFalse(self.log.exists())
|
||||
|
||||
def test_no_new_privileges_is_reported_before_start(self):
|
||||
self.status.write_text("NoNewPrivs:\t1\n")
|
||||
self.env["TEST_NO_NEW_PRIVS"] = "1"
|
||||
result = self.run_entrypoint(1)
|
||||
self.assertIn("no-new-privileges", result.stderr)
|
||||
self.assertFalse(self.conf.exists())
|
||||
|
||||
def test_missing_proc_status_field_does_not_block_startup(self):
|
||||
self.status.write_text("Name:\tmariadbd\nUid:\t999\t999\t999\t999\n")
|
||||
self.run_entrypoint()
|
||||
self.assertIn("delegate mariadbd", self.log.read_text())
|
||||
|
||||
def test_failed_privilege_query_is_not_ignored(self):
|
||||
self.env["TEST_SETPRIV_RC"] = "1"
|
||||
result = self.run_entrypoint(1)
|
||||
self.assertIn("privilege constraints", result.stderr)
|
||||
self.assertFalse(self.conf.exists())
|
||||
|
||||
def test_malformed_privilege_query_is_not_ignored(self):
|
||||
self.env["TEST_NO_NEW_PRIVS"] = "unknown"
|
||||
result = self.run_entrypoint(1)
|
||||
self.assertIn("cannot determine", result.stderr)
|
||||
self.assertFalse(self.conf.exists())
|
||||
|
||||
def test_certificate_group_is_created_when_missing(self):
|
||||
self.env["TEST_HAVE_GROUP"] = "0"
|
||||
self.run_entrypoint()
|
||||
self.assertIn("groupadd -g 12002 hostcerts", self.log.read_text())
|
||||
self.assertIn("usermod -aG hostcerts mysql", self.log.read_text())
|
||||
|
||||
def test_invalid_login_attribute_and_gid_are_rejected(self):
|
||||
for key, value in (("LDAP_LOGIN_ATTR", "uid)(objectClass)=*"), ("CERTS_GID", "not-a-gid")):
|
||||
with self.subTest(key=key):
|
||||
previous = self.env[key]
|
||||
self.env[key] = value
|
||||
self.run_entrypoint(1)
|
||||
self.env[key] = previous
|
||||
self.assertFalse(self.log.exists())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user