"""An in-memory DirectoryProtocol for tests. Passwords here are obviously fake fixture values.""" from __future__ import annotations import uuid from dataclasses import dataclass, field from packetstation.auth.groups import group_names from packetstation.auth.ldap import AdUser, LdapUnavailable FAKE_PASSWORD = "fake-password-1" NESTED_GROUP = "Shop-Welders-Nested" BASE_DN = "DC=pstest,DC=local" OU_DN = "OU=PacketStation,OU=Groups,OU=PESCO_FARMINGTON," + BASE_DN OTHER_DN = "OU=Groups,OU=PESCO_FARMINGTON," + BASE_DN @dataclass class FakeUser: sam: str display: str emp_no: str | None groups: list[str] password: str = FAKE_PASSWORD disabled: bool = False guid: uuid.UUID = field(default_factory=uuid.uuid4) @property def dn(self) -> str: return f"CN={self.display},OU=Users,OU=PESCO_FARMINGTON,DC=pstest,DC=local" @property def upn(self) -> str: return f"{self.sam}@pstest.local" def default_users() -> list[FakeUser]: """The same people as the CI domain controller (deploy/samba).""" return [ FakeUser("admin.ann", "Ann Admin", "1001", ["PS-Admins"]), FakeUser("clerk.clara", "Clara Clerk", "4001", ["PS-Clerical"]), FakeUser("clerk.leo", "Leo Lead", "4002", ["PS-Clerical-Supervisor"]), FakeUser("wh.wade", "Wade Warehouse", "6001", ["PS-Warehouse"]), FakeUser("qc.quinn", "Quinn QC", "5001", ["PS-QC"]), FakeUser("pm.paula", "Paula PM", "7001", ["PS-ProjectMgmt"]), FakeUser("weld.andy", "Andy Welder", "3003", ["PS-03-WeldParts-Worker"]), FakeUser("lead.lou", "Lou Lead", "3002", ["PS-03-WeldParts-Lead"]), FakeUser("sup.wes", "Wes Super", "2003", ["PS-03-WeldParts-Supervisor"]), FakeUser("multi.mia", "Mia Multi", "2006", ["PS-Clerical", "PS-06-CodeBldg2-Supervisor"]), FakeUser("nested.ned", "Ned Nested", "3010", [NESTED_GROUP]), FakeUser("both.bo", "Bo Both", "3011", ["PS-03-WeldParts-Worker", "PS-03-WeldParts-Supervisor"]), FakeUser("disabled.dan", "Dan Disabled", "3004", ["PS-03-WeldParts-Worker"], disabled=True), FakeUser("noemp.nora", "Nora NoEmp", None, ["PS-Warehouse"]), FakeUser("dup.dave", "Dave Dup", "9001", ["PS-03-WeldParts-Worker"]), FakeUser("dup.dina", "Dina Dup", "9001", ["PS-Warehouse"]), FakeUser("norole.nick", "Nick NoRole", "8001", []), ] class FakeDirectory: """Users, groups in the PacketStation OU, and nested groups outside it.""" def __init__( self, users: list[FakeUser] | None = None, ou_groups: list[str] | None = None, nested: dict[str, list[str]] | None = None, other_groups: dict[str, str] | None = None, ): self.users = users if users is not None else default_users() if ou_groups is None: ou_groups = [g for g in group_names() if g != "PS-14-Electrical-Lead"] + ["PS-Extra-Legacy"] self.ou_groups = ou_groups self.nested = nested if nested is not None else {NESTED_GROUP: ["PS-10-WeldAssembly-Worker"]} # Groups outside the OU, by name, with a description (the nested helper group and a mail group) self.other_groups = ( other_groups if other_groups is not None else {NESTED_GROUP: "Helper group for the nested membership test", "IT-Helpdesk": "IT help desk"} ) self.down = False self.binds = 0 self.searches = 0 # ------------------------------------------------ helpers def user(self, sam: str) -> FakeUser: return next(u for u in self.users if u.sam == sam) def _check(self) -> None: if self.down: raise LdapUnavailable("fake directory is down") self.searches += 1 def _ad(self, u: FakeUser) -> AdUser: return AdUser(u.guid, u.dn, u.sam, u.upn, u.display, u.emp_no, f"{u.sam}@example.test", u.disabled) def _closure(self, u: FakeUser) -> set[str]: seen: set[str] = set() todo = list(u.groups) while todo: g = todo.pop() if g.lower() in seen: continue seen.add(g.lower()) todo.extend(self.nested.get(g, [])) return seen def group_dn(self, name: str) -> str: """The DN of a group: in the PacketStation OU, or under Groups for the ones outside it.""" if any(g.lower() == name.lower() for g in self.ou_groups): return f"CN={name},{OU_DN}" return f"CN={name},{OTHER_DN}" def _group_by_dn(self, dn: str) -> str | None: for name in [*self.ou_groups, *self.other_groups]: if self.group_dn(name).lower() == dn.strip().lower(): return name return None def _in_ou(self, names: set[str]) -> list[str]: return sorted(g for g in self.ou_groups if g.lower() in names) # ------------------------------------------------ protocol def find_user(self, login: str) -> AdUser | None: self._check() key = login.strip().lower() for u in self.users: if key in (u.sam.lower(), u.upn.lower()): return self._ad(u) return None def get_user_by_guid(self, guid: uuid.UUID) -> AdUser | None: self._check() for u in self.users: if u.guid == guid: return self._ad(u) return None def verify_password(self, dn: str, password: str) -> bool: if self.down: raise LdapUnavailable("fake directory is down") if not password: return False self.binds += 1 return any(u.dn == dn and u.password == password for u in self.users) def groups_of(self, dn: str) -> list[str]: self._check() u = next((x for x in self.users if x.dn == dn), None) return [] if u is None else self._in_ou(self._closure(u)) def list_groups(self) -> list[str]: self._check() return sorted(self.ou_groups) def members_of(self, group_cn: str) -> list[AdUser]: self._check() if group_cn.lower() not in {g.lower() for g in self.ou_groups}: return [] return [self._ad(u) for u in self.users if group_cn.lower() in self._closure(u)] def list_groups_detail(self) -> list[tuple[str, str]]: self._check() return sorted(((g, self.group_dn(g)) for g in self.ou_groups), key=lambda t: t[0].lower()) def members_of_dn(self, group_dn: str) -> list[AdUser]: self._check() name = self._group_by_dn(group_dn) if name is None: return [] return [self._ad(u) for u in self.users if name.lower() in self._closure(u)] def search_groups(self, q: str, limit: int = 25) -> list[dict[str, str]]: self._check() key = q.strip().lower() if not key: return [] out = [] for name in sorted({*self.ou_groups, *self.other_groups}, key=str.lower): desc = self.other_groups.get(name, f"Packet Station: {name}") if key in name.lower() or key in desc.lower(): out.append({"dn": self.group_dn(name), "name": name, "description": desc}) return out[: max(1, limit)] def users_by_guid(self, guids: list[uuid.UUID]) -> dict[uuid.UUID, AdUser | None]: self._check() by = {u.guid: u for u in self.users} return {g: (self._ad(by[g]) if g in by else None) for g in guids}