diff --git a/README.md b/README.md
index 130ab94..640e1b3 100644
--- a/README.md
+++ b/README.md
@@ -197,7 +197,7 @@ Export to CSV:
- `--userenum`: Extract usernames via CUCM User Data Services (UDS) API (paginates the full directory) and harvest the full directory records (names incl. nickname, phone/home/mobile/pager numbers, email, directory URI, MS URI, department, title, manager, UUID) into the `uds_directory` table; always writes `cucm_directory.csv` (override with `--directory-outfile`)
- `--directory`: Harvest the unauthenticated CUCM corporate directory from `/cucm-uds/users` without any device probing or config downloads — requires `-H`; always writes `cucm_directory.csv` (override with `--directory-outfile`), prints a console table, and stores to `uds_directory` unless `--no-db`
- `--directory-outfile FILENAME`: Override the default CSV output path for `--directory` and `--userenum` (default: `cucm_directory.csv`)
-- `--servers`: Enumerate CUCM cluster members (hostnames + IPs) via UDS `/cucm-uds/servers` — requires `-H`
+- `--servers`: Enumerate CUCM cluster members via UDS `/cucm-uds/servers` — requires `-H`. UDS returns a hostname per cluster member and nothing else, so addresses and Publisher/Subscriber roles are not available from this endpoint
- `--http`: Use HTTP (port 6970) as the primary config download protocol with TFTP fallback (default: TFTP first, HTTP fallback)
- `--uds-port PORT`: Override the CUCM UDS API HTTPS port for `--userenum`, `--directory`, and `--servers` (default: 8443)
- `--spray`: Password-spray the UDS API (requires `-H`; mutually exclusive with `--brute-mac`)
diff --git a/src/seeyoucm_thief/thief.py b/src/seeyoucm_thief/thief.py
index 32ecc4a..b105550 100644
--- a/src/seeyoucm_thief/thief.py
+++ b/src/seeyoucm_thief/thief.py
@@ -386,8 +386,9 @@ def get_version(cucm_host, port=UDS_PORT, timeout=10):
if not cucm_host:
return None
if _TEST_MODE:
- return {'version': '12.5.1-TEST', 'prefix': '11.0(1)',
- 'usersAuthRequired': False, 'port': port}
+ return {'version': '12.5.1-TEST', 'schemaVersion': '10.0.0',
+ 'usersAuthRequired': False, 'upgradeInProgress': False,
+ 'port': port}
url = f'https://{cucm_host}:{port}/cucm-uds/version'
dbg(f'UDS GET {url} (timeout={timeout}s)')
@@ -401,11 +402,25 @@ def get_version(cucm_host, port=UDS_PORT, timeout=10):
dbg(f'UDS version non-200 body (first 300 chars): {resp.text[:300]!r}')
return None
+ # Per the v14 schema (version.get.xsd) the whole response is:
+ # <- version attr = UDS schema
+ # 14.0.1 <- element = CUCM version
+ #
+ # bool
+ # bool
+ #
+ #
+ # There are no other elements; is mandatory on 14 but absent
+ # before 11.5(1), so both flags stay unset on older clusters.
info = {}
- for field in ('version', 'prefix'):
- m = re.search(rf'<{field}>([^<]+)', resp.text)
- if m:
- info[field] = m.group(1).strip()
+ m = re.search(r'([^<]+)', resp.text)
+ if m:
+ info['version'] = m.group(1).strip()
+ # The version= attribute on the wrapper is the UDS schema version, which is
+ # distinct from the CUCM version in the element.
+ m = re.search(r']*\bversion="([^"]+)"', resp.text)
+ if m:
+ info['schemaVersion'] = m.group(1).strip()
# UDS advertises whether /users needs Basic auth via this flag (present
# since 11.5). When true, unauthenticated --userenum/--directory/--spray
# will come back empty even though the version probe succeeded, so surface
@@ -414,6 +429,12 @@ def get_version(cucm_host, port=UDS_PORT, timeout=10):
resp.text, re.IGNORECASE)
if auth_m:
info['usersAuthRequired'] = auth_m.group(1).lower() == 'true'
+ # A cluster mid-upgrade can return partial or inconsistent directory data,
+ # so results captured now should not be trusted as a complete baseline.
+ upg_m = re.search(r'\s*(true|false)\s*',
+ resp.text, re.IGNORECASE)
+ if upg_m:
+ info['upgradeInProgress'] = upg_m.group(1).lower() == 'true'
if info:
info['port'] = port
return info or None
@@ -562,11 +583,36 @@ def _add(name):
return filenames
+def _parse_uds_users_wrapper(body):
+ """Extract the paging attributes from the wrapper element.
+
+ The v14 UDS schema (users.get.xsd) declares start, requestedCount,
+ returnedCount and totalCount as use="required", so a conformant response
+ always carries all four. Returns a dict of the ones actually present, so a
+ non-conformant server degrades to the -counting fallback rather
+ than raising."""
+ m = re.search(r']*)>', body)
+ if not m:
+ return {}
+ attrs = m.group(1)
+ found = {}
+ for name in ('start', 'requestedCount', 'returnedCount', 'totalCount'):
+ am = re.search(rf'\b{name}="(-?\d+)"', attrs)
+ if am:
+ found[name] = int(am.group(1))
+ return found
+
+
def _iter_uds_user_pages(cucm_host, port=UDS_PORT, timeout=10, max_pages=10000):
"""Yield each /cucm-uds/users page's response body text, following UDS
pagination. Shared by get_users_api and get_user_directory_api so the
- page-walking logic lives in one place. Pagination 'item count' is measured by
- occurrences (UDS paginates per user)."""
+ page-walking logic lives in one place.
+
+ Paging is driven by the wrapper's own start/returnedCount
+ attributes: the next offset is start + returnedCount by construction, which
+ avoids having to assume whether UDS indexes from 0 or 1 (it is 0-based). If
+ a server omits those attributes, fall back to counting
+ occurrences and offsetting by the running total."""
base = f'https://{cucm_host}:{port}/cucm-uds/users'
seen_urls = set()
next_url = base
@@ -598,20 +644,37 @@ def _iter_uds_user_pages(cucm_host, port=UDS_PORT, timeout=10, max_pages=10000):
dbg(f'UDS empty page body (first 300 chars): {resp.text[:300]!r}')
break
+ wrapper = _parse_uds_users_wrapper(resp.text)
+
yield resp.text
collected += page_count
- if total is None:
- total_match = re.search(r']*\btotalCount="(\d+)"', resp.text) \
- or re.search(r'(\d+)', resp.text)
- if total_match:
- total = int(total_match.group(1))
- dbg(f'UDS server reports totalCount={total}')
+ if total is None and 'totalCount' in wrapper:
+ total = wrapper['totalCount']
+ dbg(f'UDS server reports totalCount={total}')
if total is not None and collected >= total:
break
- next_url = _uds_next_link(resp.text, base, collected + 1)
+ # Prefer the server's own offset arithmetic. requestedCount >
+ # returnedCount means the server clamped the page below what was asked
+ # for (UserSearchLimit), which is normal and not an error.
+ if 'start' in wrapper and 'returnedCount' in wrapper:
+ if wrapper['returnedCount'] <= 0:
+ dbg('UDS returnedCount=0; stopping pagination')
+ break
+ requested = wrapper.get('requestedCount')
+ if requested is not None and requested > wrapper['returnedCount']:
+ dbg(f'UDS clamped page size: requested {requested}, '
+ f'returned {wrapper["returnedCount"]}')
+ next_start = wrapper['start'] + wrapper['returnedCount']
+ else:
+ # Non-conformant server: offset by what we have actually collected.
+ # start is 0-based, so the next page begins at `collected`, not
+ # `collected + 1`.
+ next_start = collected
+
+ next_url = _uds_next_link(base, next_start)
if not next_url:
dbg('UDS no next-page link found; stopping pagination')
break
@@ -665,7 +728,11 @@ def parse_uds_directory(xml_body):
continue
record = {'username': name_match.group(1).strip()}
for key, tag in _UDS_DIRECTORY_FIELDS:
- m = re.search(rf'<{tag}>([^<]*)', block)
+ # Tolerate attributes on the element: users.get.xsd gives
+ # an optional exist="true|false" attribute, and a
+ # bare match silently yields '' when CUCM sets it. \b keeps
+ # from matching a longer tag that merely starts with "id".
+ m = re.search(rf'<{tag}\b[^>]*>([^<]*)', block)
record[key] = m.group(1).strip() if m else ''
records.append(record)
return records
@@ -691,7 +758,7 @@ def get_user_directory_api(cucm_host, port=UDS_PORT, timeout=10, max_pages=10000
def get_servers_api(cucm_host, port=UDS_PORT, timeout=10):
if _TEST_MODE:
- return [{'hostName': 'cucm-pub.test', 'ipv4Address': '10.0.0.1', 'serverType': 'Publisher'}]
+ return [{'hostName': 'cucm-pub.test'}, {'hostName': 'cucm-sub1.test'}]
url = f'https://{cucm_host}:{port}/cucm-uds/servers'
dbg(f'UDS GET {url} (timeout={timeout}s)')
@@ -705,36 +772,31 @@ def get_servers_api(cucm_host, port=UDS_PORT, timeout=10):
dbg(f'UDS servers non-200 body (first 300 chars): {resp.text[:300]!r}')
return []
+ # The v14 schema (servers.get.xsd) declares as type="xs:string" —
+ # a bare hostname with no child elements, and the schema permits none. This
+ # is the only shape UDS has ever published (10.0(1) through 14), so there is
+ # no version branching to do here. UDS exposes no addresses and no server
+ # role: resolving a hostname to an IP is a DNS lookup on our side, and
+ # Publisher/Subscriber role requires AXL.
servers = []
for block in re.findall(r']*>(.*?)', resp.text, re.DOTALL):
- srv = {}
- for field in ('hostName', 'ipv4Address', 'ipv6Address', 'serverType'):
- m = re.search(rf'<{field}>([^<]+)', block)
- if m:
- srv[field] = m.group(1).strip()
- if not srv:
- # Newer UDS (15.x+) returns the hostname as plain text inside
- # ... with no child elements.
- text = re.sub(r'<[^>]+>', '', block).strip()
- if text:
- srv['hostName'] = text
- if srv:
- servers.append(srv)
+ host = block.strip()
+ if host:
+ servers.append({'hostName': host})
dbg(f'UDS parsed {len(servers)} server entries from cluster topology')
return servers
-def _uds_next_link(body, base_url, fallback_start):
- # HATEOAS variants seen in CUCM UDS responses
- m = re.search(r']*\brel="next"[^>]*\bhref="([^"]+)"', body, re.IGNORECASE)
- if m:
- return m.group(1)
- m = re.search(r'([^<]+)', body, re.IGNORECASE)
- if m:
- return m.group(1).strip()
- # Fallback: try ?start=N (most common Cisco UDS pagination param)
+def _uds_next_link(base_url, start):
+ """Build the URL for the UDS /users page beginning at offset ``start``.
+
+ UDS exposes no HATEOAS paging links — the v14 schema's wrapper
+ permits only the uri/version/start/requestedCount/returnedCount/totalCount
+ attributes and children, with no or
+ element in any published release. Client-side offset arithmetic on the
+ 0-based ``start`` parameter is the only paging mechanism available."""
sep = '&' if '?' in base_url else '?'
- return f'{base_url}{sep}start={fallback_start}'
+ return f'{base_url}{sep}start={start}'
def log_uds_usernames_to_db(cucm_host, usernames, db_file='thief.db'):
@@ -2033,6 +2095,11 @@ def log_cluster_servers_to_db(queried_host, servers, db_file='thief.db'):
inserted = 0
for srv in servers:
hostname = srv.get('hostName') or ''
+ # UDS /cucm-uds/servers returns only a hostname per (v14
+ # schema: type="xs:string"). ipv4/ipv6/server_type are retained as
+ # columns for a future DNS-resolution or AXL enrichment step and for
+ # compatibility with existing thief.db files, but nothing UDS
+ # returns can populate them.
ipv4 = srv.get('ipv4Address') or ''
ipv6 = srv.get('ipv6Address') or ''
server_type = srv.get('serverType') or ''
@@ -2420,10 +2487,13 @@ def display_database_summary(db_file='thief.db', cucm_filter=None):
if cluster_servers:
print(f'\n\033[1m[+] CUCM CLUSTER SERVERS ({len(cluster_servers)} total)\033[0m')
print("-"*70)
- print(f'{"Queried Host":<24} {"Hostname":<30} {"IPv4":<16}')
+ # No address column: UDS /cucm-uds/servers returns only a hostname
+ # per cluster member, so ipv4/ipv6/server_type are always empty for
+ # rows this tool writes (see log_cluster_servers_to_db).
+ print(f'{"Queried Host":<24} {"Hostname":<30} {"Discovered":<20}')
print("-"*70)
for queried, hostname, ipv4, ipv6, srv_type, timestamp in cluster_servers:
- print(f'{queried:<24} {(hostname or ""):<30} {(ipv4 or ""):<16}')
+ print(f'{queried:<24} {(hostname or ""):<30} {(timestamp or ""):<20}')
if spray_hits:
print('\n=== UDS Spray Hits ===')
@@ -2929,10 +2999,14 @@ def main():
if version_info:
uds_port = version_info.get('port', args.uds_port)
v = version_info.get('version', 'unknown')
- p = version_info.get('prefix')
- print(f'[+] CUCM {CUCM_host} version: {v}' + (f' (prefix {p})' if p else ''))
+ sv = version_info.get('schemaVersion')
+ print(f'[+] CUCM {CUCM_host} version: {v}' + (f' (UDS schema {sv})' if sv else ''))
if uds_port != args.uds_port:
print(f'[*] UDS answered on :{uds_port} (Contact Search Authentication likely enabled); using it for this run')
+ if version_info.get('upgradeInProgress'):
+ print('[!] CUCM reports an upgrade in progress (upgradeInProgress=true).')
+ print(' UDS results may be incomplete or inconsistent; treat this run as '
+ 'provisional and re-enumerate once the upgrade finishes.')
if version_info.get('usersAuthRequired'):
print('[!] This cluster requires authentication for the UDS /users resource '
'(usersResourceAuthEnabled=true).')
@@ -2952,18 +3026,7 @@ def main():
quit(0)
print(f'[+] Discovered {len(servers)} cluster member(s):')
for srv in servers:
- host = srv.get('hostName', '?')
- ipv4 = srv.get('ipv4Address', '')
- ipv6 = srv.get('ipv6Address', '')
- srv_type = srv.get('serverType', '')
- parts = [host]
- if ipv4:
- parts.append(f'({ipv4})')
- if ipv6:
- parts.append(f'[v6: {ipv6}]')
- if srv_type:
- parts.append(f'<{srv_type}>')
- print(' ' + ' '.join(parts))
+ print(' ' + srv.get('hostName', '?'))
if not no_db:
inserted = log_cluster_servers_to_db(CUCM_host, servers, db_file)
print(f'[+] Logged {inserted} new cluster server entry/entries to database')
diff --git a/tests/test_brute_host_prefixes.py b/tests/test_brute_host_prefixes.py
index 301f507..1ac7321 100644
--- a/tests/test_brute_host_prefixes.py
+++ b/tests/test_brute_host_prefixes.py
@@ -88,7 +88,7 @@ def test_servers_feature_still_probes_uds_version(monkeypatch, tmp_path):
"""UDS features must keep running the version probe."""
calls = []
monkeypatch.setattr(thief, 'get_version',
- lambda *a, **kw: calls.append(kw) or {'version': '14.0', 'prefix': None})
+ lambda *a, **kw: calls.append(kw) or {'version': '14.0'})
monkeypatch.setattr(thief, 'get_servers_api', lambda *a, **kw: [])
db_file = str(tmp_path / 'thief.db')
thief.init_database(db_file)
diff --git a/tests/test_spray.py b/tests/test_spray.py
index c06394f..26cbe99 100644
--- a/tests/test_spray.py
+++ b/tests/test_spray.py
@@ -528,7 +528,7 @@ def fake_run_spray(**kwargs):
called.update(kwargs)
monkeypatch.setattr(thief, 'run_spray', fake_run_spray)
- monkeypatch.setattr(thief, 'get_version', lambda *a, **kw: {'version': '15.0', 'prefix': None})
+ monkeypatch.setattr(thief, 'get_version', lambda *a, **kw: {'version': '15.0'})
db = tmp_path / "thief.db"
monkeypatch.setattr('sys.argv', [
'thief', '-H', '1.2.3.4', '--spray',
@@ -557,7 +557,7 @@ def fake_run_spray(**kwargs):
called.update(kwargs)
monkeypatch.setattr(thief, 'run_spray', fake_run_spray)
- monkeypatch.setattr(thief, 'get_version', lambda *a, **kw: {'version': '15.0', 'prefix': None})
+ monkeypatch.setattr(thief, 'get_version', lambda *a, **kw: {'version': '15.0'})
db = tmp_path / "thief.db"
monkeypatch.setattr('sys.argv', [
'thief', '-H', '1.2.3.4', '--spray', '-P', str(pw_file),
diff --git a/tests/test_uds_hardening.py b/tests/test_uds_hardening.py
index 0ed381f..52e0ec1 100644
--- a/tests/test_uds_hardening.py
+++ b/tests/test_uds_hardening.py
@@ -13,10 +13,17 @@ def _disable_test_mode(monkeypatch):
def _version_xml(auth=None):
- body = "14.0.114.0(1)"
+ """Shaped per the official v14 version.get.xsd: a
+ wrapper whose version= attribute is the UDS schema version, a
+ element holding the CUCM version, and the capability flags nested under
+ ."""
+ caps = ''
if auth is not None:
- body += f"{'true' if auth else 'false'}"
- return body + ""
+ caps = (''
+ f"{'true' if auth else 'false'}"
+ '')
+ return (''
+ f'14.0.1{caps}')
def _resp(text, status=200):
diff --git a/tests/test_uds_schema_conformance.py b/tests/test_uds_schema_conformance.py
new file mode 100644
index 0000000..5106a29
--- /dev/null
+++ b/tests/test_uds_schema_conformance.py
@@ -0,0 +1,369 @@
+"""Tests pinning UDS parsing to the official Cisco v14 XML schemas
+(UDS-xsd-14.zip, developer.cisco.com/site/user-data-services/downloads/schemas/).
+
+The schemas are the authoritative source here — Cisco's prose docs stop at
+12.5(1) and disagree with themselves in places. Element/attribute shapes
+asserted below are quoted from the XSDs in the docstrings.
+"""
+import re
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from seeyoucm_thief import thief
+
+
+@pytest.fixture(autouse=True)
+def _disable_test_mode(monkeypatch):
+ monkeypatch.setattr(thief, '_TEST_MODE', False)
+
+
+def _resp(text, status=200):
+ r = MagicMock()
+ r.status_code = status
+ r.text = text
+ r.content = text.encode()
+ return r
+
+
+def _users_page(users, start, total, requested=None, returned=None):
+ """A schema-conformant page.
+
+ users.get.xsd declares start/requestedCount/returnedCount/totalCount as
+ use="required" on the wrapper.
+ """
+ returned = len(users) if returned is None else returned
+ requested = returned if requested is None else requested
+ body = ''.join(
+ f'id-{u}'
+ f'{u}'
+ for u in users
+ )
+ return (f'{body}')
+
+
+# ---------------------------------------------------------------------------
+# Pagination (#33): start is 0-based; drive paging off the wrapper attributes
+# ---------------------------------------------------------------------------
+
+def test_next_page_offset_comes_from_start_plus_returned_count():
+ """A full first page of 64 must be followed by start=64, not start=65."""
+ page1 = _users_page([f'u{i}' for i in range(64)], start=0, total=100)
+ page2 = _users_page([f'u{i}' for i in range(64, 100)], start=64, total=100)
+ urls = []
+
+ def side_effect(url, **kwargs):
+ urls.append(url)
+ return _resp(page1 if len(urls) == 1 else page2)
+
+ with patch.object(thief.requests, 'get', side_effect=side_effect):
+ users = thief.get_users_api('cucm.example.com', 8443)
+
+ assert urls[1] == 'https://cucm.example.com:8443/cucm-uds/users?start=64'
+ assert len(users) == 100
+
+
+def test_no_user_is_skipped_across_the_page_boundary():
+ """The off-by-one in #33 silently dropped the first user of every page
+ after the first.
+
+ The fake server honours the requested ``start`` offset by slicing the
+ directory, exactly as a real CUCM does. That is what makes this a real
+ regression test: an off-by-one in the requested offset makes the server
+ genuinely skip a user, so the assertion on the collected set fails on its
+ own rather than relying on the URL assertions below.
+ """
+ expected = [f'user{i:03d}' for i in range(130)]
+ page_size = 64
+ seen = []
+
+ def side_effect(url, **kwargs):
+ seen.append(url)
+ m = re.search(r'[?&]start=(\d+)', url)
+ start = int(m.group(1)) if m else 0
+ window = expected[start:start + page_size]
+ return _resp(_users_page(window, start=start, total=len(expected),
+ requested=page_size))
+
+ with patch.object(thief.requests, 'get', side_effect=side_effect):
+ users = thief.get_users_api('cucm.example.com', 8443)
+
+ assert users == expected
+ assert seen[1].endswith('?start=64')
+ assert seen[2].endswith('?start=128')
+
+
+def test_pagination_tolerates_a_server_clamping_the_page_size():
+ """requestedCount > returnedCount means UserSearchLimit clamped the page.
+ That is normal; paging must follow returnedCount, not requestedCount."""
+ page1 = _users_page(['a', 'b'], start=0, total=4, requested=64, returned=2)
+ page2 = _users_page(['c', 'd'], start=2, total=4, requested=64, returned=2)
+ urls = []
+
+ def side_effect(url, **kwargs):
+ urls.append(url)
+ return _resp(page1 if len(urls) == 1 else page2)
+
+ with patch.object(thief.requests, 'get', side_effect=side_effect):
+ users = thief.get_users_api('cucm.example.com', 8443)
+
+ assert urls[1].endswith('?start=2')
+ assert users == ['a', 'b', 'c', 'd']
+
+
+def test_pagination_falls_back_to_collected_count_without_wrapper_attrs():
+ """A non-conformant server that omits start/returnedCount still pages, and
+ the fallback offset is 0-based (collected, not collected + 1)."""
+ page1 = 'a' \
+ 'b'
+ page2 = 'c'
+ urls = []
+
+ def side_effect(url, **kwargs):
+ urls.append(url)
+ return _resp(page1 if len(urls) == 1 else page2)
+
+ with patch.object(thief.requests, 'get', side_effect=side_effect):
+ users = thief.get_users_api('cucm.example.com', 8443)
+
+ assert urls[1].endswith('?start=2')
+ assert users == ['a', 'b', 'c']
+
+
+def test_pagination_stops_when_returned_count_is_zero():
+ page1 = _users_page(['a'], start=0, total=99, returned=0)
+ calls = []
+
+ def side_effect(url, **kwargs):
+ calls.append(url)
+ return _resp(page1)
+
+ with patch.object(thief.requests, 'get', side_effect=side_effect):
+ users = thief.get_users_api('cucm.example.com', 8443)
+
+ assert len(calls) == 1
+ assert users == ['a']
+
+
+def test_uds_next_link_appends_start_to_a_bare_base():
+ assert thief._uds_next_link('https://c:8443/cucm-uds/users', 64) == \
+ 'https://c:8443/cucm-uds/users?start=64'
+
+
+def test_uds_next_link_uses_ampersand_when_query_already_present():
+ assert thief._uds_next_link('https://c:8443/cucm-uds/users?max=500', 64) == \
+ 'https://c:8443/cucm-uds/users?max=500&start=64'
+
+
+def test_parse_uds_users_wrapper_reads_all_four_required_attrs():
+ w = thief._parse_uds_users_wrapper(_users_page(['a'], start=8, total=9))
+ assert w == {'start': 8, 'requestedCount': 1, 'returnedCount': 1,
+ 'totalCount': 9}
+
+
+def test_parse_uds_users_wrapper_returns_empty_without_a_wrapper():
+ assert thief._parse_uds_users_wrapper('') == {}
+
+
+# ---------------------------------------------------------------------------
+# /cucm-uds/servers (#36): is type="xs:string"
+# ---------------------------------------------------------------------------
+
+SERVERS_XML = (
+ ''
+ 'cucm-pub.example.com'
+ 'cucm-sub1.example.com'
+ '10.0.0.9'
+ ''
+)
+
+
+def test_servers_parsed_as_plain_text_hostnames(monkeypatch):
+ """servers.get.xsd: — a bare
+ hostname, no child elements, on every release 10.0(1) through 14."""
+ monkeypatch.setattr(thief.requests, 'get', lambda *a, **kw: _resp(SERVERS_XML))
+ servers = thief.get_servers_api('cucm', port=8443)
+ assert servers == [
+ {'hostName': 'cucm-pub.example.com'},
+ {'hostName': 'cucm-sub1.example.com'},
+ {'hostName': '10.0.0.9'},
+ ]
+
+
+def test_servers_parse_tolerates_pretty_printed_whitespace(monkeypatch):
+ xml = ('\n'
+ ' \n cucm-pub.example.com\n \n'
+ '\n')
+ monkeypatch.setattr(thief.requests, 'get', lambda *a, **kw: _resp(xml))
+ assert thief.get_servers_api('cucm', port=8443) == \
+ [{'hostName': 'cucm-pub.example.com'}]
+
+
+def test_servers_parse_skips_empty_server_elements(monkeypatch):
+ xml = (''
+ 'real.example.com')
+ monkeypatch.setattr(thief.requests, 'get', lambda *a, **kw: _resp(xml))
+ assert thief.get_servers_api('cucm', port=8443) == \
+ [{'hostName': 'real.example.com'}]
+
+
+def test_servers_result_never_carries_axl_style_fields(monkeypatch):
+ """hostName/ipv4Address/ipv6Address/serverType were AXL vocabulary that UDS
+ never returns. Only hostName should ever be present."""
+ monkeypatch.setattr(thief.requests, 'get', lambda *a, **kw: _resp(SERVERS_XML))
+ for srv in thief.get_servers_api('cucm', port=8443):
+ assert set(srv) == {'hostName'}
+
+
+def test_servers_empty_on_non_200(monkeypatch):
+ monkeypatch.setattr(thief.requests, 'get',
+ lambda *a, **kw: _resp('nope', status=401))
+ assert thief.get_servers_api('cucm', port=8443) == []
+
+
+# ---------------------------------------------------------------------------
+# /cucm-uds/version (#35, #37)
+# ---------------------------------------------------------------------------
+
+def _version_xml(auth=None, upgrading=None, schema='10.0.0', cucm='14.0.1'):
+ """A schema-conformant version.get.xsd response."""
+ caps = ''
+ if auth is not None:
+ caps += f"{'true' if auth else 'false'}"
+ if upgrading is not None:
+ caps += f"{'true' if upgrading else 'false'}"
+ if caps:
+ caps = f'{caps}'
+ return (f'{cucm}{caps}'
+ f'')
+
+
+def test_version_parse_has_no_prefix_key(monkeypatch):
+ """version.get.xsd permits only and ; there is no
+ element on any release."""
+ monkeypatch.setattr(thief.requests, 'get',
+ lambda *a, **kw: _resp(_version_xml(auth=False)))
+ info = thief.get_version('cucm', port=8443)
+ assert 'prefix' not in info
+ assert info['version'] == '14.0.1'
+
+
+def test_version_parses_schema_version_from_wrapper_attribute(monkeypatch):
+ """The version= attribute is the UDS schema version and is distinct from
+ the element's CUCM version."""
+ monkeypatch.setattr(
+ thief.requests, 'get',
+ lambda *a, **kw: _resp(_version_xml(auth=False, schema='10.0.0', cucm='14.0.1')))
+ info = thief.get_version('cucm', port=8443)
+ assert info['schemaVersion'] == '10.0.0'
+ assert info['version'] == '14.0.1'
+
+
+@pytest.mark.parametrize('flag', [True, False])
+def test_version_parses_upgrade_in_progress(monkeypatch, flag):
+ monkeypatch.setattr(thief.requests, 'get',
+ lambda *a, **kw: _resp(_version_xml(auth=False, upgrading=flag)))
+ assert thief.get_version('cucm', port=8443)['upgradeInProgress'] is flag
+
+
+def test_version_omits_upgrade_flag_when_capabilities_absent(monkeypatch):
+ """ is mandatory in v14 but absent before 11.5(1). A missing
+ flag must stay absent rather than defaulting to a value."""
+ monkeypatch.setattr(thief.requests, 'get', lambda *a, **kw: _resp(_version_xml()))
+ info = thief.get_version('cucm', port=8443)
+ assert 'upgradeInProgress' not in info
+ assert 'usersAuthRequired' not in info
+
+
+def test_version_parses_both_capability_flags_together(monkeypatch):
+ monkeypatch.setattr(
+ thief.requests, 'get',
+ lambda *a, **kw: _resp(_version_xml(auth=True, upgrading=True)))
+ info = thief.get_version('cucm', port=8443)
+ assert info['usersAuthRequired'] is True
+ assert info['upgradeInProgress'] is True
+
+
+# ---------------------------------------------------------------------------
+# main() wiring for the upgrade warning (#37)
+# ---------------------------------------------------------------------------
+
+def test_main_warns_when_upgrade_in_progress(monkeypatch, tmp_path, capsys):
+ monkeypatch.setattr(thief, 'probe_uds',
+ lambda *a, **kw: {'version': '14.0.1', 'port': 8443,
+ 'upgradeInProgress': True})
+ monkeypatch.setattr(thief, 'get_users_api', lambda *a, **kw: [])
+ db = tmp_path / 'thief.db'
+ thief.init_database(str(db))
+ monkeypatch.setattr('sys.argv',
+ ['thief', '--userenum', '-H', 'cucm', '--db', str(db)])
+ with pytest.raises(SystemExit):
+ thief.main()
+ out = capsys.readouterr().out
+ assert 'upgradeInProgress=true' in out
+
+
+def test_main_silent_when_no_upgrade_in_progress(monkeypatch, tmp_path, capsys):
+ monkeypatch.setattr(thief, 'probe_uds',
+ lambda *a, **kw: {'version': '14.0.1', 'port': 8443,
+ 'upgradeInProgress': False})
+ monkeypatch.setattr(thief, 'get_users_api', lambda *a, **kw: [])
+ db = tmp_path / 'thief.db'
+ thief.init_database(str(db))
+ monkeypatch.setattr('sys.argv',
+ ['thief', '--userenum', '-H', 'cucm', '--db', str(db)])
+ with pytest.raises(SystemExit):
+ thief.main()
+ assert 'upgradeInProgress' not in capsys.readouterr().out
+
+
+def test_main_reports_uds_schema_version(monkeypatch, tmp_path, capsys):
+ monkeypatch.setattr(thief, 'probe_uds',
+ lambda *a, **kw: {'version': '14.0.1', 'port': 8443,
+ 'schemaVersion': '10.0.0'})
+ monkeypatch.setattr(thief, 'get_users_api', lambda *a, **kw: [])
+ db = tmp_path / 'thief.db'
+ thief.init_database(str(db))
+ monkeypatch.setattr('sys.argv',
+ ['thief', '--userenum', '-H', 'cucm', '--db', str(db)])
+ with pytest.raises(SystemExit):
+ thief.main()
+ assert 'UDS schema 10.0.0' in capsys.readouterr().out
+
+
+# ---------------------------------------------------------------------------
+# parse_uds_directory: elements may legally carry attributes
+# ---------------------------------------------------------------------------
+
+def test_directory_uri_parsed_when_exist_attribute_present():
+ """users.get.xsd gives an optional exist="true|false"
+ attribute. A bare-tag regex silently yielded '' whenever CUCM set it,
+ dropping the SIP/Jabber URI from every harvested record."""
+ xml = (''
+ ''
+ 'uuid-alicealice'
+ 'alice@corp.example'
+ '')
+ records = thief.parse_uds_directory(xml)
+ assert records[0]['directory_uri'] == 'alice@corp.example'
+
+
+def test_directory_uri_still_parsed_without_attributes():
+ xml = ('bob'
+ 'bob@corp.example')
+ assert thief.parse_uds_directory(xml)[0]['directory_uri'] == 'bob@corp.example'
+
+
+def test_directory_user_id_not_confused_by_a_longer_tag():
+ """The pattern must not match a tag that merely starts with "id"."""
+ xml = (''
+ 'not-the-id'
+ 'caroluuid-carol'
+ '')
+ assert thief.parse_uds_directory(xml)[0]['user_id'] == 'uuid-carol'