Compare commits

..

2 Commits

5 changed files with 169 additions and 35 deletions

View File

@ -18,6 +18,34 @@ CLI tool to query LDAP/AD servers
* By default the first 1000 entries are shown, and it errors if there are more results * By default the first 1000 entries are shown, and it errors if there are more results
* Use `--all` to show all results * Use `--all` to show all results
## Virtual attributes
`ldaptool` supports constructing new values from existing attributes by adding a `:<postprocess>` suffix (which can be chained apart from the length limit).
* Some suffixes support an argument as `:<postprocess>[<arg>]`.
* A single integer as postprocess suffix limits the length of the value; it replaces the last character of the output with `…` if it cut something off.
* Multi-valued attributes generate multiple virtual attrites; each value is processed individually. (The values are joined afterwards for table output if needed.)
### DN handling
DNs are decoded into lists of lists of `(name, value)` pairs (the inner list usually contains exactly one entry).
Attributes with a `DC` name are considered part of the "domain", everything else belongs to the "path".
(Usually a DN will start with path segments and end with domain segments.)
The path is read from back to front.
The following postprocess hooks are available:
* `domain`: extracts the domain as DNS FQDN (`CN=Someone,OU=Dep1,DC=example,DC=com` becomes `example.com`)
* `path`: extracts the non-domain parts without names and separates them by `/` (`CN=Someone,OU=Dep1,DC=example,DC=com` becomes `Dep1/Someone`)
* `fullpath`: uses the `domain` as first segment in a path (`CN=Someone,OU=Dep1,DC=example,DC=com` becomes `example.com/Dep1/Someone`)
* `dnslice`: extracts a "slice" from a DN (outer list only); the result is still in DN format.
`path`, `fullpath` and `dnslice` take an optional index/slice as argument, written in python syntax.
For `path` and `fullpath` this extracts only the given index/slice from the path (`fullpath` always includes the full FQDN as first segment), `dnslice` operates on the outer list of decoded (lists of) pairs:
* `dn:dnslice[1:]` on `dn: CN=Someone,OU=Dep1,DC=example,DC=com` returns `OU=Dep1,DC=example,DC=com`
* `dn:fullpath[:-1]` on `dn: CN=Someone,OU=Dep1,DC=example,DC=com` returns `example.com/Dep1`
* `dn:path[-1]` on `dn: CN=Someone,OU=Dep1,DC=example,DC=com` returns `Someone`
## Authentication, Protocol, Ports ## Authentication, Protocol, Ports
`ldaptool` always uses TLS for password based authentication, and SASL GSS-API over non-TLS for Kerberos ones. `ldaptool` always uses TLS for password based authentication, and SASL GSS-API over non-TLS for Kerberos ones.

View File

@ -33,19 +33,26 @@ class DNInfo:
def domain(self) -> str: def domain(self) -> str:
return ".".join(ava[1] for rdn in self.parts for ava in rdn if ava[0].lower() == "dc") return ".".join(ava[1] for rdn in self.parts for ava in rdn if ava[0].lower() == "dc")
def _path(self, *, escape: typing.Callable[[str], str], sep: str) -> str: def _path(self, *, escape: typing.Callable[[str], str], sep: str, selection: slice = slice(None)) -> str:
return sep.join(escape(ava[1]) for rdn in reversed(self.parts) for ava in rdn if ava[0].lower() != "dc") rev_flattened = [ava[1] for rdn in reversed(self.parts) for ava in rdn if ava[0].lower() != "dc"]
return sep.join(value for value in rev_flattened[selection])
def sliced_path(self, selection: slice, /) -> str:
return self._path(escape=lambda value: _escape_backslash(value, special="/"), sep="/", selection=selection)
@functools.cached_property @functools.cached_property
def path(self) -> str: def path(self) -> str:
return self._path(escape=lambda value: _escape_backslash(value, special="/"), sep="/") return self.sliced_path(slice(None))
@property def sliced_full_path(self, selection: slice, /) -> str:
def full_path(self) -> str:
domain = self.domain domain = self.domain
path = self.path path = self.sliced_path(selection)
if not path: if not path:
return self.domain return self.domain
if not domain: if not domain:
return self.path return self.path
return f"{domain}/{path}" return f"{domain}/{path}"
@property
def full_path(self) -> str:
return self.sliced_full_path(slice(None))

View File

@ -101,7 +101,7 @@ class Attribute:
return return
def _try_decode(self, args: Arguments) -> None: def _try_decode(self, args: Arguments) -> None:
if self.name in ("objectSid",): if self.name in ("objectSid","securityIdentifier"):
self._try_decode_sid() self._try_decode_sid()
elif self.name in ("msExchMailboxGuid", "objectGUID"): elif self.name in ("msExchMailboxGuid", "objectGUID"):
self._try_decode_uuid() self._try_decode_uuid()

View File

@ -2,6 +2,9 @@ from __future__ import annotations
import abc import abc
import dataclasses import dataclasses
import typing
import ldap.dn
from ldaptool._utils.dninfo import DNInfo from ldaptool._utils.dninfo import DNInfo
@ -14,6 +17,27 @@ class Step(abc.ABC):
... ...
def _args_to_slice(args: str) -> slice:
args = args.strip()
if not args:
return slice(None)
params: list[typing.Optional[int]] = []
for arg in args.split(":"):
arg = arg.strip()
if arg:
params.append(int(arg))
else:
params.append(None)
if len(params) == 1:
assert isinstance(params[0], int)
ndx = params[0]
if ndx == -1:
return slice(ndx, None) # from last element to end - still exactly one element
# this doesn't work for ndx == -1: slice(-1, 0) is always empty. otherwise it should return [ndx:][:1].
return slice(ndx, ndx + 1)
return slice(*params)
@dataclasses.dataclass(slots=True) @dataclasses.dataclass(slots=True)
class MaxLength(Step): class MaxLength(Step):
limit: int limit: int
@ -26,6 +50,10 @@ class MaxLength(Step):
@dataclasses.dataclass(slots=True) @dataclasses.dataclass(slots=True)
class DNDomain(Step): class DNDomain(Step):
def __init__(self, args: str) -> None:
if args:
raise ValueError(":domain doesn't support an argument")
def step(self, value: str) -> str: def step(self, value: str) -> str:
try: try:
dninfo = DNInfo(dn=value) dninfo = DNInfo(dn=value)
@ -37,30 +65,57 @@ class DNDomain(Step):
@dataclasses.dataclass(slots=True) @dataclasses.dataclass(slots=True)
class DNPath(Step): class DNPath(Step):
path_slice: slice
def __init__(self, args: str) -> None:
self.path_slice = _args_to_slice(args)
def step(self, value: str) -> str: def step(self, value: str) -> str:
try: try:
dninfo = DNInfo(dn=value) dninfo = DNInfo(dn=value)
except Exception: except Exception:
# not a valid DN -> no processing # not a valid DN -> no processing
return value return value
return dninfo.path return dninfo.sliced_path(self.path_slice)
@dataclasses.dataclass(slots=True) @dataclasses.dataclass(slots=True)
class DNFullPath(Step): class DNFullPath(Step):
path_slice: slice
def __init__(self, args: str) -> None:
self.path_slice = _args_to_slice(args)
def step(self, value: str) -> str: def step(self, value: str) -> str:
try: try:
dninfo = DNInfo(dn=value) dninfo = DNInfo(dn=value)
except Exception: except Exception:
# not a valid DN -> no processing # not a valid DN -> no processing
return value return value
return dninfo.full_path return dninfo.sliced_full_path(self.path_slice)
_STEPS = { @dataclasses.dataclass(slots=True)
"domain": DNDomain(), class DNSlice(Step):
"path": DNPath(), slice: slice
"fullpath": DNFullPath(),
def __init__(self, args: str) -> None:
self.slice = _args_to_slice(args)
def step(self, value: str) -> str:
try:
dninfo = DNInfo(dn=value)
except Exception:
# not a valid DN -> no processing
return value
return ldap.dn.dn2str(dninfo.parts[self.slice]) # type: ignore
_STEPS: dict[str, typing.Callable[[str], Step]] = {
"domain": DNDomain,
"path": DNPath,
"fullpath": DNFullPath,
"dnslice": DNSlice,
} }
@ -78,19 +133,63 @@ class PostProcess:
return value return value
def parse_steps(steps: list[str]) -> PostProcess: def parse_steps(steps: str) -> PostProcess:
max_len = 0 result: list[Step] = []
try:
max_len = int(steps[-1]) cur_id_start = 0
steps.pop() cur_args_start = -1
except ValueError: current_id = ""
pass current_args = ""
result = [] count_brackets = 0
for step in steps: step_done = False
step_i = _STEPS.get(step, None)
def handle_step() -> None:
nonlocal cur_id_start, cur_args_start, current_id, current_args, step_done
assert step_done
step_i = _STEPS.get(current_id, None)
if step_i is None: if step_i is None:
raise InvalidStep(f"Unknown post-processing step {step!r}") try:
result.append(step_i) max_len = int(current_id)
if max_len:
result.append(MaxLength(max_len)) result.append(MaxLength(max_len))
except ValueError:
raise InvalidStep(f"Unknown post-processing step {current_id!r}")
else:
result.append(step_i(current_args))
cur_id_start = pos + 1
cur_args_start = -1
current_id = ""
current_args = ""
step_done = False
for pos, char in enumerate(steps):
if step_done:
if char != ":":
raise InvalidStep(f"Require : after step, found {char!r} at pos {pos}")
handle_step()
elif char == "[":
if count_brackets == 0:
# end of identifier
current_id = steps[cur_id_start:pos]
cur_args_start = pos + 1
count_brackets += 1
elif char == "]":
count_brackets -= 1
if count_brackets == 0:
current_args = steps[cur_args_start:pos]
step_done = True
elif count_brackets:
continue
elif not char.isalnum():
raise InvalidStep(f"Expecting either alphanumeric, ':' or '[', got {char!r} at {pos}")
if not step_done:
current_id = steps[cur_id_start:]
if current_id:
step_done = True
if step_done:
handle_step()
return PostProcess(result) return PostProcess(result)

View File

@ -57,20 +57,20 @@ class Arguments(argclasses.BaseArguments):
self.columns_keys.append(column) self.columns_keys.append(column)
if column == "dndomain": if column == "dndomain":
self.post_process.setdefault("dn", {})[column] = _postprocess.parse_steps(["domain"]) self.post_process.setdefault("dn", {})[column] = _postprocess.parse_steps("domain")
attributes_set.add("dn") attributes_set.add("dn")
elif column == "dnpath": elif column == "dnpath":
self.post_process.setdefault("dn", {})[column] = _postprocess.parse_steps(["path"]) self.post_process.setdefault("dn", {})[column] = _postprocess.parse_steps("path")
attributes_set.add("dn") attributes_set.add("dn")
elif column == "dnfullpath": elif column == "dnfullpath":
self.post_process.setdefault("dn", {})[column] = _postprocess.parse_steps(["fullpath"]) self.post_process.setdefault("dn", {})[column] = _postprocess.parse_steps("fullpath")
attributes_set.add("dn") attributes_set.add("dn")
else: else:
step_names = column.split(":") col_parts = column.split(":", maxsplit=1)
attributes_set.add(step_names[0]) attributes_set.add(col_parts[0])
if len(step_names) > 1: if len(col_parts) == 2:
source = step_names.pop(0) source, steps = col_parts
self.post_process.setdefault(source, {})[column] = _postprocess.parse_steps(step_names) self.post_process.setdefault(source, {})[column] = _postprocess.parse_steps(steps)
if all_attributes: if all_attributes:
self.attributes = [] self.attributes = []