ebbe.bassandClaude Sonnet 5.5
700295cb72
Split load_config into per-section readers, use contextlib.suppress and a single condition where ruff asked, remove stale noqa comments, add the SIM and RUF100 rules, share the sign_in test helper, align AD-6 with the code and record the remaining deferred findings. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N3t4WL7bd9bdHHNjEsxzqC
115 lines
2.9 KiB
Python
115 lines
2.9 KiB
Python
"""A stand-in for the `ldap` module that records calls. Real python-ldap needs a C library."""
|
|
|
|
from types import SimpleNamespace
|
|
|
|
|
|
class LDAPError(Exception):
|
|
pass
|
|
|
|
|
|
class SERVER_DOWN(LDAPError):
|
|
pass
|
|
|
|
|
|
class INVALID_CREDENTIALS(LDAPError):
|
|
pass
|
|
|
|
|
|
class INSUFFICIENT_ACCESS(LDAPError):
|
|
pass
|
|
|
|
|
|
class CONSTRAINT_VIOLATION(LDAPError):
|
|
pass
|
|
|
|
|
|
class UNWILLING_TO_PERFORM(LDAPError):
|
|
pass
|
|
|
|
|
|
class INVALID_SYNTAX(LDAPError):
|
|
pass
|
|
|
|
|
|
def escape_filter_chars(value):
|
|
table = {"\\": r"\5c", "*": r"\2a", "(": r"\28", ")": r"\29", "\x00": r"\00"}
|
|
return "".join(table.get(c, c) for c in value)
|
|
|
|
|
|
class Connection:
|
|
def __init__(self, world, url):
|
|
self.world = world
|
|
self.url = url
|
|
self.options = {}
|
|
self.bound = None
|
|
|
|
def set_option(self, name, value):
|
|
self.options[name] = value
|
|
|
|
def simple_bind_s(self, dn, password):
|
|
if self.world.down:
|
|
raise SERVER_DOWN()
|
|
if self.world.passwords.get(dn) != password:
|
|
raise INVALID_CREDENTIALS()
|
|
self.bound = dn
|
|
|
|
def unbind_s(self):
|
|
pass
|
|
|
|
def search_s(self, base, scope, flt, attrs):
|
|
self.world.searches.append((base, scope, flt, tuple(attrs)))
|
|
return self.world.respond(base, scope, flt)
|
|
|
|
def modify_s(self, dn, mods):
|
|
self.world.modifications.append((dn, mods))
|
|
if self.world.modify_error:
|
|
raise self.world.modify_error
|
|
|
|
|
|
class World:
|
|
"""The server side: canned search answers and the connections the adapter opened."""
|
|
|
|
def __init__(self):
|
|
self.down = False
|
|
self.passwords = {"svc@example.test": "service-password"}
|
|
self.searches = []
|
|
self.modifications = []
|
|
self.modify_error = None
|
|
self.connections = []
|
|
self.entries = {} # (base, scope) -> list of (dn, attrs)
|
|
|
|
def respond(self, base, scope, flt):
|
|
return self.entries.get((base, scope), [])
|
|
|
|
|
|
def build(world):
|
|
def initialize(url):
|
|
conn = Connection(world, url)
|
|
world.connections.append(conn)
|
|
return conn
|
|
|
|
return SimpleNamespace(
|
|
initialize=initialize,
|
|
LDAPError=LDAPError,
|
|
SERVER_DOWN=SERVER_DOWN,
|
|
INVALID_CREDENTIALS=INVALID_CREDENTIALS,
|
|
INSUFFICIENT_ACCESS=INSUFFICIENT_ACCESS,
|
|
CONSTRAINT_VIOLATION=CONSTRAINT_VIOLATION,
|
|
UNWILLING_TO_PERFORM=UNWILLING_TO_PERFORM,
|
|
INVALID_SYNTAX=INVALID_SYNTAX,
|
|
filter=SimpleNamespace(escape_filter_chars=escape_filter_chars),
|
|
SCOPE_BASE=0,
|
|
SCOPE_SUBTREE=2,
|
|
MOD_ADD=0,
|
|
MOD_DELETE=1,
|
|
MOD_REPLACE=2,
|
|
OPT_PROTOCOL_VERSION="protocol",
|
|
OPT_REFERRALS="referrals",
|
|
OPT_NETWORK_TIMEOUT="network_timeout",
|
|
OPT_TIMEOUT="timeout",
|
|
OPT_X_TLS_CACERTFILE="cacertfile",
|
|
OPT_X_TLS_REQUIRE_CERT="require_cert",
|
|
OPT_X_TLS_DEMAND="demand",
|
|
OPT_X_TLS_NEWCTX="newctx",
|
|
)
|