Run tests with docker container so we don't need to install vault

Use LDAP server in docker as well
Configure Vault/LDAP with terraform
Added LDAP tests - not yet finished

tests/config_files/vault-ldap/docker-compose.yml

@@ -0,0 +1,39 @@
+version: "3.9"
+
+services:
+ vault:
+ image: hashicorp/vault:1.15.1
+ #command: sh
+ command:
+ - vault
+ - server
+ - -config
+ - /hvac_conf/vault-tls.hcl
+ environment:
+ VAULT_ADDR: 'https://0.0.0.0:8200'
+ VAULT_API_ADDR: 'https://127.0.0.1:8200'
+ VAULT_SKIP_VERIFY: true
+ cap_add:
+ - IPC_LOCK
+ volumes:
+ - ../:/hvac_conf
+ - ../../:/tests
+ healthcheck:
+ test: "vault status; test $? -eq 2"
+ interval: 1s
+ timeout: 3s
+ retries: 5
+ start_period: 5s
+ ports:
+ - "8200-8201:8200-8201"
+
+ openldap:
+ image: bitnami/openldap:2.6
+ ports:
+ - '1389:1389'
+ - '1636:1636'
+ environment:
+ - LDAP_ADMIN_USERNAME=admin
+ - LDAP_ADMIN_PASSWORD=adminpassword
+ - LDAP_USERS=vaulttest
+ - LDAP_PASSWORDS=vaulttestpassword

tests/config_files/vault-ldap/terraform.tf

@@ -0,0 +1,53 @@
+terraform {
+ required_providers {
+ vault = {
+ source = "hashicorp/vault"
+ version = "3.22.0"
+ }
+ }
+}
+
+variable "token" {
+ type = string
+ default = "root"
+}
+
+provider "vault" {
+ address = "https://localhost:8200"
+ token = var.token
+ skip_tls_verify = true
+}
+
+resource "vault_ldap_secret_backend" "config" {
+ path = "ldap"
+ binddn = "CN=admin,dc=example,dc=org"
+ bindpass = "adminpassword"
+ url = "ldap:https://openldap:1389"
+ insecure_tls = "true"
+ userdn = "dc=example,dc=org"
+}
+
+resource "vault_ldap_secret_backend_dynamic_role" "vault-dynamic" {
+ mount = vault_ldap_secret_backend.config.path
+ role_name = "vault-dynamic"
+ creation_ldif = <<EOT
+dn: cn={{.Username}},ou=users,dc=example,dc=org
+objectClass: person
+objectClass: top
+cn: learn
+sn: {{.Password | utf16le | base64}}
+userPassword: {{.Password}}
+EOT
+ deletion_ldif = <<EOT
+dn: cn={{.Username}},ou=users,dc=example,dc=org
+changetype: delete
+EOT
+}
+
+resource "vault_ldap_secret_backend_static_role" "vault-static" {
+ mount = vault_ldap_secret_backend.config.path
+ username = "vaulttest"
+ dn = "cn=vaulttest,ou=users,dc=example,dc=org"
+ role_name = "vault-static"
+ rotation_period = 600
+}

tests/config_files/vault-tls.hcl

@@ -2,8 +2,10 @@ backend "inmem" {
 }
 
 listener "tcp" {
- tls_cert_file = "tests/config_files/server-cert.pem"
- tls_key_file = "tests/config_files/server-key.pem"
+ tls_cert_file = "/tests/config_files/server-cert.pem"
+ tls_key_file = "/tests/config_files/server-key.pem"
+ tls_min_version = "tls10"
+ address = "0.0.0.0:8200"
 }
 
 disable_mlock = true

tests/conftest.py

@@ -0,0 +1,26 @@
+import pytest
+import distutils
+import subprocess
+
+from tests.utils import get_config_file_path
+
+@pytest.fixture(scope="class", autouse=True)
+def test_setup(request):
+ def _test_setup():
+ if distutils.spawn.find_executable("docker") and distutils.spawn.find_executable("terraform"):
+ print("Setting up docker/terraform")
+ # We can spin up Vault inside docker and use terraform to do some base configuration
+ docker_file = get_config_file_path("vault-ldap/docker-compose.yml")
+ try:
+ subprocess.check_call(f"docker compose -f '{docker_file}' up -d --wait vault openldap", shell=True)
+ except Exception as e:
+ print("Failed to setup docker/terraform for test, you must have Vault installed locally. Error: " + str(e))
+
+ def _test_teardown():
+ print("Tearing down docker/terraform")
+ docker_file = get_config_file_path("vault-ldap/docker-compose.yml")
+ subprocess.check_call(f"docker compose -f '{docker_file}' down vault openldap", shell=True)
+
+ _test_teardown()
+ _test_setup()
+ request.addfinalizer(_test_teardown)

tests/integration_tests/api/secrets_engines/test_ldap.py

@@ -0,0 +1,199 @@
+from unittest import TestCase
+
+from parameterized import parameterized
+
+from hvac import Client, exceptions
+from tests.utils.hvac_integration_test_case import HvacIntegrationTestCase
+
+
+class TestLDAP(HvacIntegrationTestCase, TestCase):
+ DEFAULT_MOUNT_POINT = "ldap"
+
+ def setUp(self):
+ super().setUp()
+ self.client = Client(
+ url="http:https://localhost:8200",
+ token="root"
+ )
+
+ def tearDown(self):
+ super().tearDown()
+
+ @parameterized.expand(
+ [
+ ("CN=admin,DC=example,DC=org", "adminpassword", "ldap:https://openldap:1389", None, None, None, None, None, DEFAULT_MOUNT_POINT)
+ ]
+ )
+ def test_configure(
+ self,
+ binddn=None,
+ bindpass=None,
+ url=None,
+ password_policy=None,
+ schema=None,
+ userdn=None,
+ userattr=None,
+ upndomain=None,
+ mount_point=DEFAULT_MOUNT_POINT,
+ raises=None,
+ exception_message="",
+ ):
+ if raises:
+ with self.assertRaises(raises) as cm:
+ self.client.secrets.ldap.configure(
+ binddn, bindpass, url, password_policy, schema, 
+ userdn, userattr, upndomain, mount_point)
+
+ self.assertIn(
+ member=exception_message,
+ container=str(cm.exception),
+ )
+ else:
+ configure_result = self.client.secrets.ldap.configure(
+ binddn, bindpass, url, password_policy, schema, 
+ userdn, userattr, upndomain, mount_point)
+ self.assertEqual(
+ first=204,
+ second=configure_result.status_code,
+ )
+ print(str(configure_result))
+ # Ensure when we query it's the same
+ read_config_result = self.client.secrets.ldap.read_config(mount_point=mount_point)
+ self.assertEqual(
+ first=url,
+ second=read_config_result["data"]["url"]
+ )
+ self.assertEqual(
+ first=binddn,
+ second=read_config_result["data"]["binddn"]
+ )
+
+ @parameterized.expand(
+ [
+ ("invalid", exceptions.InvalidPath),
+ (DEFAULT_MOUNT_POINT),
+ ]
+ )
+ def test_rotate_root(
+ self, 
+ mount_point=DEFAULT_MOUNT_POINT,
+ raises=None, 
+ exception_message=""
+ ):
+ if raises:
+ with self.assertRaises(raises) as cm:
+ self.client.secrets.ldap.rotate_root(mount_point)
+
+ self.assertIn(
+ member=exception_message,
+ container=str(cm.exception),
+ )
+ else:
+ rotate_result = self.client.secrets.ldap.rotate_root(mount_point)
+ self.assertEqual(
+ first=204,
+ second=rotate_result.status_code,
+ )
+
+ @parameterized.expand(
+ [
+ ("vault-static", "vaulttest", "cn=vaulttest,ou=users,dc=example,dc=org", "24h", DEFAULT_MOUNT_POINT),
+ ("vault-static-already-managed", "vaulttest", None, None, DEFAULT_MOUNT_POINT, exceptions.InvalidRequest),
+ ]
+ )
+ def test_create_or_update_static_role(
+ self,
+ name,
+ username=None,
+ dn=None,
+ rotation_period=None,
+ mount_point=DEFAULT_MOUNT_POINT,
+ raises=None, 
+ exception_message=""
+ ):
+ if raises:
+ with self.assertRaises(raises) as cm:
+ self.client.secrets.ldap.create_or_update_static_role(name, username, dn, rotation_period, mount_point=mount_point)
+
+ self.assertIn(
+ member=exception_message,
+ container=str(cm.exception),
+ )
+ else:
+ static_result = self.client.secrets.ldap.create_or_update_static_role(name, username, dn, rotation_period, mount_point=mount_point)
+ self.assertEqual(
+ first=204,
+ second=static_result.status_code,
+ )
+
+ @parameterized.expand(
+ [
+ ("vault-static", DEFAULT_MOUNT_POINT, False),
+ ("vault-static-already-managed", DEFAULT_MOUNT_POINT, False, exceptions.InvalidPath),
+ ]
+ )
+ def test_read_static_role(
+ self, 
+ name, 
+ mount_point=DEFAULT_MOUNT_POINT,
+ create_role_before_test=True,
+ raises=None, 
+ exception_message=""
+ ):
+ username="vaulttest"
+ dn="cn=vaulttest,ou=users,dc=example,dc=org"
+ rotation_period=86400
+ if create_role_before_test:
+ self.client.secrets.ldap.create_or_update_static_role(name, username, dn, rotation_period, mount_point=mount_point)
+
+ if raises:
+ with self.assertRaises(raises) as cm:
+ self.client.secrets.ldap.read_static_role(name, mount_point=mount_point)
+
+ self.assertIn(
+ member=exception_message,
+ container=str(cm.exception),
+ )
+ else:
+ static_result = self.client.secrets.ldap.read_static_role(name, mount_point=mount_point)
+ self.assertEqual(
+ first=username,
+ second=static_result["data"]["username"],
+ )
+ self.assertEqual(
+ first=dn,
+ second=static_result["data"]["dn"],
+ )
+ self.assertEqual(
+ first=rotation_period,
+ second=static_result["data"]["rotation_period"],
+ )
+
+ def test_list_static_roles(
+ self, 
+ mount_point=DEFAULT_MOUNT_POINT,
+ create_role_before_test=True,
+ raises=None, 
+ exception_message=""
+ ):
+ pass
+
+ def test_delete_static_role(
+ self, 
+ name, 
+ mount_point=DEFAULT_MOUNT_POINT,
+ create_role_before_test=True,
+ raises=None, 
+ exception_message=""
+ ):
+ pass
+
+ def test_generate_static_credentials(
+ self, 
+ name, 
+ mount_point=DEFAULT_MOUNT_POINT,
+ create_role_before_test=True,
+ raises=None, 
+ exception_message=""
+ ):
+ pass

tests/scripts/destroy_vault_ldap.sh

@@ -0,0 +1,3 @@
+terraform -chdir="`dirname $0`"/../config_files/vault-ldap init || :
+terraform -chdir="`dirname $0`"/../config_files/vault-ldap destroy -auto-approve || :
+docker compose -f "`dirname $0`"/../config_files/vault-ldap/docker-compose.yml down vault openldap

tests/scripts/init_vault_ldap.sh

@@ -0,0 +1,3 @@
+docker compose -f "`dirname $0`"/../config_files/vault-ldap/docker-compose.yml up -d vault openldap
+terraform -chdir="`dirname $0`"/../config_files/vault-ldap init
+terraform -chdir="`dirname $0`"/../config_files/vault-ldap apply -auto-approve

tests/utils/__init__.py

@@ -18,13 +18,29 @@
 VERSION_REGEX = re.compile(r"Vault v([0-9.]+)")
 LATEST_VAULT_VERSION = "1.1.3"
 
+def vault_running_inside_docker():
+ command = ["docker", "ps"]
+ process = subprocess.Popen(**get_popen_kwargs(args=command, stdout=subprocess.PIPE))
+ output, _ = process.communicate()
+ lines = output.strip().split('\n')
+ for l in lines:
+ if 'hashicorp/vault' in l:
+ return l.split()[0].strip()
+
+ return None
 
 def get_vault_version_string():
  if "cache" in get_vault_version_string.__dict__:
  return get_vault_version_string.cache
- if not find_executable("vault"):
- raise SkipTest("Vault executable not found")
+
  command = ["vault", "-version"]
+ container = vault_running_inside_docker()
+ if not container:
+ if not find_executable("vault"):
+ raise SkipTest("Vault executable not found")
+ else:
+ command = ["docker", "exec", container] + command
+
  process = subprocess.Popen(**get_popen_kwargs(args=command, stdout=subprocess.PIPE))
  output, _ = process.communicate()
  version_string = output.strip().split()[1].lstrip("v")
@@ -185,6 +201,10 @@ def decode_generated_root_token(encoded_token, otp):
  otp,
  ]
  )
+
+ container = vault_running_inside_docker()
+ if container:
+ command = ["docker", "exec", container] + command
  process = subprocess.Popen(
  **get_popen_kwargs(args=command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  )

tests/utils/hvac_integration_test_case.py

@@ -44,7 +44,8 @@ def setUpClass(cls):
  cls.manager.start()
  cls.manager.initialize()
  cls.manager.unseal()
- except Exception:
+ cls.manager.configure()
+ except Exception as e:
  cls.manager.stop()
  raise