test: perform the ldapsearch in integration tests

.github/workflows/tests.yaml

@@ -36,7 +36,7 @@ jobs:
           bootstrap-options: '--agent-version=3.4.0'
 
       - name: Run integration tests
-        run: tox -e integration -- --model testing
+        run: tox -e build-prerequisites,integration -- --model testing
 
       - name: Get contexts
         run: kubectl config view

.pre-commit-config.yaml

@@ -32,6 +32,7 @@ repos:
       args: ["--config-file", "pyproject.toml"]
       additional_dependencies:
         - types-PyYAML
+        - types-aiofiles
 - repo: https://github.com/compilerla/conventional-pre-commit
   rev: v3.2.0
   hooks:

integration-requirements.txt

@@ -1,4 +1,7 @@
 -r requirements.txt
+aiofiles
 juju
+psycopg[binary]
 pytest
 pytest-operator
+python-ldap

tests/integration/conftest.py

@@ -2,16 +2,21 @@
 # See LICENSE file for licensing details.
 
 import functools
-import subprocess
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Callable, Optional
 
-import pytest
+import aiofiles
+import ldap.ldapobject
+import psycopg
+import pytest_asyncio
 import yaml
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from pytest_operator.plugin import OpsTest
 
+from constants import GLAUTH_LDAP_PORT
+
 METADATA = yaml.safe_load(Path("./charmcraft.yaml").read_text())
 CERTIFICATE_PROVIDER_APP = "self-signed-certificates"
 DB_APP = "postgresql-k8s"
@@ -20,6 +25,16 @@
 GLAUTH_CLIENT_APP = "any-charm"
 
 
+@contextmanager
+def ldap_connection(uri: str, bind_dn: str, bind_password: str) -> ldap.ldapobject.LDAPObject:
+    conn = ldap.initialize(uri)
+    try:
+        conn.simple_bind_s(bind_dn, bind_password)
+        yield conn
+    finally:
+        conn.unbind_s()
+
+
 def extract_certificate_common_name(certificate: str) -> Optional[str]:
     cert_data = certificate.encode()
     cert = x509.load_pem_x509_certificate(cert_data, default_backend())
@@ -29,58 +44,130 @@ def extract_certificate_common_name(certificate: str) -> Optional[str]:
     return rdns[0].rfc4514_string()
 
 
-def get_unit_data(unit_name: str, model_name: str) -> dict:
-    res = subprocess.run(
-        ["juju", "show-unit", unit_name, "-m", model_name],
-        check=True,
-        text=True,
-        capture_output=True,
-    )
-    cmd_output = yaml.safe_load(res.stdout)
+async def get_secret(ops_test: OpsTest, secret_id: str) -> dict:
+    show_secret_cmd = f"show-secret {secret_id} --reveal".split()
+    _, stdout, _ = await ops_test.juju(*show_secret_cmd)
+    cmd_output = yaml.safe_load(stdout)
+    return cmd_output[secret_id]
+
+
+async def get_unit_data(ops_test: OpsTest, unit_name: str) -> dict:
+    show_unit_cmd = f"show-unit {unit_name}".split()
+    _, stdout, _ = await ops_test.juju(*show_unit_cmd)
+    cmd_output = yaml.safe_load(stdout)
     return cmd_output[unit_name]
 
 
-def get_integration_data(model_name: str, app_name: str, integration_name: str) -> Optional[dict]:
-    unit_data = get_unit_data(f"{app_name}/0", model_name)
+async def get_integration_data(
+    ops_test: OpsTest, app_name: str, integration_name: str, unit_num: int = 0
+) -> Optional[dict]:
+    data = await get_unit_data(ops_test, f"{app_name}/{unit_num}")
     return next(
         (
             integration
-            for integration in unit_data["relation-info"]
+            for integration in data["relation-info"]
             if integration["endpoint"] == integration_name
         ),
         None,
     )
 
 
-def get_app_integration_data(
-    model_name: str, app_name: str, integration_name: str
+async def get_app_integration_data(
+    ops_test: OpsTest,
+    app_name: str,
+    integration_name: str,
+    unit_num: int = 0,
 ) -> Optional[dict]:
-    data = get_integration_data(model_name, app_name, integration_name)
+    data = await get_integration_data(ops_test, app_name, integration_name, unit_num)
     return data["application-data"] if data else None
 
 
-def get_unit_integration_data(
-    model_name: str, app_name: str, remote_app_name: str, integration_name: str
+async def get_unit_integration_data(
+    ops_test: OpsTest,
+    app_name: str,
+    remote_app_name: str,
+    integration_name: str,
 ) -> Optional[dict]:
-    data = get_integration_data(model_name, app_name, integration_name)
+    data = await get_integration_data(ops_test, app_name, integration_name)
     return data["related-units"][f"{remote_app_name}/0"]["data"] if data else None
 
 
-@pytest.fixture
-def app_integration_data(ops_test: OpsTest) -> Callable:
-    return functools.partial(get_app_integration_data, ops_test.model_name)
+@pytest_asyncio.fixture
+async def app_integration_data(ops_test: OpsTest) -> Callable:
+    return functools.partial(get_app_integration_data, ops_test)
+
+
+@pytest_asyncio.fixture
+async def unit_integration_data(ops_test: OpsTest) -> Callable:
+    return functools.partial(get_unit_integration_data, ops_test)
+
+
+@pytest_asyncio.fixture
+async def ldap_integration_data(app_integration_data: Callable) -> Optional[dict]:
+    return await app_integration_data(GLAUTH_CLIENT_APP, "ldap")
+
+
+@pytest_asyncio.fixture
+async def database_integration_data(app_integration_data: Callable) -> Optional[dict]:
+    return await app_integration_data(GLAUTH_APP, "pg-database")
+
+
+@pytest_asyncio.fixture
+async def certificate_integration_data(app_integration_data: Callable) -> Optional[dict]:
+    return await app_integration_data(GLAUTH_APP, "certificates")
+
+
+@pytest_asyncio.fixture
+async def ldap_configurations(
+    ops_test: OpsTest, ldap_integration_data: Optional[dict]
+) -> Optional[tuple[str, ...]]:
+    if not ldap_integration_data:
+        return None
+
+    base_dn = ldap_integration_data["base_dn"]
+    bind_dn = ldap_integration_data["bind_dn"]
+    bind_password_secret: str = ldap_integration_data["bind_password_secret"]
+
+    prefix, _, secret_id = bind_password_secret.partition(":")
+    bind_password = await get_secret(ops_test, secret_id or prefix)
+
+    return base_dn, bind_dn, bind_password["content"]["password"]
+
+
+async def unit_address(ops_test: OpsTest, *, app_name: str, unit_num: int = 0) -> str:
+    status = await ops_test.model.get_status()
+    return status["applications"][app_name]["units"][f"{app_name}/{unit_num}"]["address"]
+
+
+@pytest_asyncio.fixture
+async def ldap_uri(ops_test: OpsTest) -> str:
+    address = await unit_address(ops_test, app_name=GLAUTH_APP)
+    return f"ldap://{address}:{GLAUTH_LDAP_PORT}"
+
 
+@pytest_asyncio.fixture
+async def database_address(ops_test: OpsTest) -> str:
+    return await unit_address(ops_test, app_name=DB_APP)
 
-@pytest.fixture
-def unit_integration_data(ops_test: OpsTest) -> Callable:
-    return functools.partial(get_unit_integration_data, ops_test.model_name)
 
+@pytest_asyncio.fixture
+async def initialize_database(
+    database_integration_data: Optional[dict], database_address: str
+) -> None:
+    assert database_integration_data is not None, "database_integration_data should be ready"
 
-@pytest.fixture
-def database_integration_data(app_integration_data: Callable) -> Optional[dict]:
-    return app_integration_data(GLAUTH_APP, "pg-database")
+    db_connection_params = {
+        "dbname": database_integration_data["database"],
+        "user": database_integration_data["username"],
+        "password": database_integration_data["password"],
+        "host": database_address,
+        "port": 5432,
+    }
 
+    async with await psycopg.AsyncConnection.connect(**db_connection_params) as conn:
+        async with conn.cursor() as cursor:
+            async with aiofiles.open("tests/integration/db.sql", "rb") as f:
+                statements = await f.read()
 
-@pytest.fixture
-def certificate_integration_data(app_integration_data: Callable) -> Optional[dict]:
-    return app_integration_data(GLAUTH_APP, "certificates")
+            await cursor.execute(statements)
+            await conn.commit()

tests/integration/db.sql

@@ -0,0 +1,18 @@
+INSERT INTO ldapgroups(name, gidnumber) VALUES('superheros', 5502);
+INSERT INTO ldapgroups(name, gidnumber) VALUES('svcaccts', 5503);
+INSERT INTO ldapgroups(name, gidnumber) VALUES('civilians', 5504);
+INSERT INTO ldapgroups(name, gidnumber) VALUES('caped', 5505);
+INSERT INTO ldapgroups(name, gidnumber) VALUES('lovesailing', 5506);
+INSERT INTO ldapgroups(name, gidnumber) VALUES('smoker', 5507);
+
+INSERT INTO includegroups(parentgroupid, includegroupid) VALUES(5504, 5502);
+INSERT INTO includegroups(parentgroupid, includegroupid) VALUES(5505, 5503);
+INSERT INTO includegroups(parentgroupid, includegroupid) VALUES(5505, 5502);
+
+INSERT INTO users(name, uidnumber, primarygroup, passsha256) VALUES ('hackers', 5002, 5502, '6478579e37aff45f013e14eeb30b3cc56c72ccdc310123bcdf53e0333e3f416a');
+INSERT INTO users(name, uidnumber, primarygroup, passsha256) VALUES('johndoe', 5003, 5503, '6478579e37aff45f013e14eeb30b3cc56c72ccdc310123bcdf53e0333e3f416a');
+INSERT INTO users(name, mail, uidnumber, primarygroup, passsha256) VALUES('serviceuser', '[email protected]', 5004, 5503, '652c7dc687d98c9889304ed2e408c74b611e86a40caa51c4b43f1dd5913c5cd0');
+INSERT INTO users(name, uidnumber, primarygroup, passsha256, othergroups, custattr) VALUES('user4', 5005, 5502, '652c7dc687d98c9889304ed2e408c74b611e86a40caa51c4b43f1dd5913c5cd0', '5505,5506', '{"employeetype":["Intern","Temp"],"employeenumber":[12345,54321]}');
+
+INSERT INTO capabilities(userid, action, object) VALUES(5002, 'search', 'ou=superheros,dc=glauth,dc=com');
+INSERT INTO capabilities(userid, action, object) VALUES(5004, 'search', '*');

tests/integration/test_charm.py

@@ -8,6 +8,7 @@
 from pathlib import Path
 from typing import Callable, Optional
 
+import ldap
 import pytest
 from conftest import (
     CERTIFICATE_PROVIDER_APP,
@@ -16,6 +17,7 @@
     GLAUTH_CLIENT_APP,
     GLAUTH_IMAGE,
     extract_certificate_common_name,
+    ldap_connection,
 )
 from pytest_operator.plugin import OpsTest
 from tester import ANY_CHARM
@@ -73,11 +75,11 @@ async def test_build_and_deploy(ops_test: OpsTest) -> None:
         apps=[CERTIFICATE_PROVIDER_APP, DB_APP, GLAUTH_CLIENT_APP, GLAUTH_APP],
         status="active",
         raise_on_blocked=False,
-        timeout=1000,
+        timeout=5 * 60,
     )
 
 
-def test_database_integration(
+async def test_database_integration(
     ops_test: OpsTest,
     database_integration_data: Optional[dict],
 ) -> None:
@@ -86,7 +88,7 @@ def test_database_integration(
     assert database_integration_data["password"]
 
 
-def test_certification_integration(
+async def test_certification_integration(
     certificate_integration_data: Optional[dict],
 ) -> None:
     assert certificate_integration_data
@@ -107,18 +109,18 @@ async def test_ldap_integration(
     await ops_test.model.wait_for_idle(
         apps=[GLAUTH_APP, GLAUTH_CLIENT_APP],
         status="active",
-        timeout=1000,
+        timeout=5 * 60,
     )
 
-    ldap_integration_data = app_integration_data(
+    integration_data = await app_integration_data(
         GLAUTH_CLIENT_APP,
         "ldap",
     )
-    assert ldap_integration_data
-    assert ldap_integration_data["bind_dn"].startswith(
+    assert integration_data
+    assert integration_data["bind_dn"].startswith(
         f"cn={GLAUTH_CLIENT_APP},ou={ops_test.model_name}"
     )
-    assert ldap_integration_data["bind_password_secret"].startswith("secret:")
+    assert integration_data["bind_password_secret"].startswith("secret:")
 
 
 async def test_certificate_transfer_integration(
@@ -130,7 +132,7 @@ async def test_certificate_transfer_integration(
         f"{GLAUTH_APP}:send-ca-cert",
     )
 
-    certificate_transfer_integration_data = unit_integration_data(
+    certificate_transfer_integration_data = await unit_integration_data(
         GLAUTH_CLIENT_APP,
         GLAUTH_APP,
         "send-ca-cert",
@@ -139,19 +141,19 @@ async def test_certificate_transfer_integration(
 
 
 async def test_glauth_scale_up(ops_test: OpsTest) -> None:
-    app, target_unit_num = ops_test.model.applications[GLAUTH_APP], 3
+    app, target_unit_num = ops_test.model.applications[GLAUTH_APP], 2
 
     await app.scale(target_unit_num)
 
     await ops_test.model.wait_for_idle(
         apps=[GLAUTH_APP],
         status="active",
-        timeout=1000,
+        timeout=5 * 60,
         wait_for_exact_units=target_unit_num,
     )
 
 
-@pytest.mark.xfail(
+@pytest.mark.skip(
     reason="cert_handler is bugged, remove this once it is fixed or when we throw it away..."
 )
 async def test_glauth_scale_down(ops_test: OpsTest) -> None:
@@ -161,5 +163,57 @@ async def test_glauth_scale_down(ops_test: OpsTest) -> None:
     await ops_test.model.wait_for_idle(
         apps=[GLAUTH_APP],
         status="active",
-        timeout=1000,
+        timeout=5 * 60,
     )
+
+
+async def test_ldap_search_operation(
+    initialize_database: None,
+    ldap_uri: str,
+    ldap_configurations: Optional[tuple[str, ...]],
+) -> None:
+    assert ldap_configurations, "LDAP configuration should be ready"
+    base_dn, bind_dn, bind_password = ldap_configurations
+
+    with ldap_connection(uri=ldap_uri, bind_dn=bind_dn, bind_password=bind_password) as conn:
+        res = conn.search_s(
+            base=base_dn,
+            scope=ldap.SCOPE_SUBTREE,
+            filterstr="(cn=hackers)",
+        )
+
+    assert res[0], "User 'hackers' can be found"
+    dn, _ = res[0]
+    assert dn == f"cn=hackers,ou=superheros,ou=users,{base_dn}"
+
+    with ldap_connection(
+        uri=ldap_uri, bind_dn=f"cn=serviceuser,ou=svcaccts,{base_dn}", bind_password="mysecret"
+    ) as conn:
+        res = conn.search_s(
+            base=base_dn,
+            scope=ldap.SCOPE_SUBTREE,
+            filterstr="(cn=johndoe)",
+        )
+
+    assert res[0], "User 'johndoe' can be found by using 'serviceuser' as bind DN"
+    dn, _ = res[0]
+    assert dn == f"cn=johndoe,ou=svcaccts,ou=users,{base_dn}"
+
+    with ldap_connection(
+        uri=ldap_uri, bind_dn=f"cn=hackers,ou=superheros,{base_dn}", bind_password="dogood"
+    ) as conn:
+        user4 = conn.search_s(
+            base=f"ou=superheros,{base_dn}", scope=ldap.SCOPE_SUBTREE, filterstr="(cn=user4)"
+        )
+
+    assert user4[0], "User 'user4' can be found by using 'hackers' as bind DN"
+    dn, _ = user4[0]
+    assert dn == f"cn=user4,ou=superheros,{base_dn}"
+
+    with (
+        ldap_connection(
+            uri=ldap_uri, bind_dn=f"cn=hackers,ou=superheros,{base_dn}", bind_password="dogood"
+        ) as conn,
+        pytest.raises(ldap.INSUFFICIENT_ACCESS),
+    ):
+        conn.search_s(base=base_dn, scope=ldap.SCOPE_SUBTREE, filterstr="(cn=user4)")