From 6e1ab002a51eb3ed9e6b37dd9a3bb0b3d15d16b6 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Wed, 29 Jul 2026 13:34:12 -0400 Subject: [PATCH 1/4] fix: scope UDS version probe to UDS features; clarify --brute-mac -H error The startup version probe hit the UDS port (8443) for every -H host, even for --brute-mac and plain config/phone scans that never touch UDS. Against a host where UDS is firewalled or not listening, those runs paid a full read timeout and printed a misleading "Could not retrieve CUCM version" error unrelated to what the user asked for. Gate the probe to the features that actually use UDS (--servers, --directory, --userenum, --spray). Separately, --brute-mac with -H and no seeded prefixes printed "You must specify at least one phone with -p (or a CUCM server with -H)", implying -H was missing when it was in fact supplied. --brute-mac never queries the server; it replays MAC prefixes already harvested by --userenum/--spray or a phone scan. The message now names the host and points at the seeding steps. Co-Authored-By: Claude Fable 5 --- src/seeyoucm_thief/thief.py | 22 +++++++++++--- tests/test_brute_host_prefixes.py | 49 +++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 4 deletions(-) diff --git a/src/seeyoucm_thief/thief.py b/src/seeyoucm_thief/thief.py index ff6d04a..0b05e66 100644 --- a/src/seeyoucm_thief/thief.py +++ b/src/seeyoucm_thief/thief.py @@ -2835,7 +2835,13 @@ def main(): ) quit(0) - if CUCM_host: + # The version probe only informs the UDS-based features; it queries the UDS + # port (8443) and has nothing to say for --brute-mac or plain config/phone + # scans. Running it unconditionally made those runs pay a full UDS read + # timeout and print a misleading "could not retrieve version" error against + # hosts where UDS is firewalled or not listening. + uds_feature = args.servers or args.directory or args.userenum or args.spray + if CUCM_host and uds_feature: version_info = get_version(CUCM_host, port=args.uds_port) if version_info: v = version_info.get('version', 'unknown') @@ -3014,9 +3020,17 @@ def main(): else: db_prefixes = all_prefixes if not db_prefixes: - print('You must specify at least one phone with -p (or a CUCM server with -H) when using --brute-mac') - if not no_db: - print(' (and no previously discovered phones were found in the database)') + if CUCM_host: + # -H was given, but --brute-mac does not itself query the + # server: it replays MAC prefixes already harvested into the + # database by --userenum/--spray or an earlier phone scan. + print(f'--brute-mac found no MAC prefixes for {CUCM_host} in the database.') + print(' Run --userenum or --spray against it first, scan a phone with -p,') + print(' or pass a phone IP directly with -p to seed prefixes.') + else: + print('You must specify at least one phone with -p (or a CUCM server with -H) when using --brute-mac') + if not no_db: + print(' (and no previously discovered phones were found in the database)') quit(1) prefix_len = 12 - brute_mac_len if prefix_len < 0: diff --git a/tests/test_brute_host_prefixes.py b/tests/test_brute_host_prefixes.py index a2cfe7e..c01ad3d 100644 --- a/tests/test_brute_host_prefixes.py +++ b/tests/test_brute_host_prefixes.py @@ -1,5 +1,7 @@ import sqlite3 +import pytest + import thief @@ -48,3 +50,50 @@ def test_brute_mac_accepts_host_without_phone(tmp_path): ) assert 'You must specify at least one phone' not in result.stdout assert 'MAC brute force mode enabled using 1 MAC prefix' in result.stdout + + +def test_brute_mac_does_not_probe_uds_version(monkeypatch, tmp_path, capsys): + """--brute-mac must not run the UDS version probe: it never touches UDS, so + a firewalled 8443 would otherwise cost a full read timeout and a misleading + 'could not retrieve version' error.""" + calls = [] + monkeypatch.setattr(thief, 'get_version', lambda *a, **kw: calls.append(kw) or None) + db_file = str(tmp_path / 'thief.db') + thief.init_database(db_file) + monkeypatch.setattr('sys.argv', + ['thief', '-b', '1', '-H', 'cucm1', '--db', db_file]) + with pytest.raises(SystemExit): + thief.main() + out = capsys.readouterr().out + assert calls == [] + assert 'Could not retrieve CUCM version' not in out + + +def test_brute_mac_empty_db_message_names_host(monkeypatch, tmp_path, capsys): + """With -H but no seeded prefixes, the error should point at seeding steps, + not claim -H was missing.""" + monkeypatch.setattr(thief, 'get_version', lambda *a, **kw: None) + db_file = str(tmp_path / 'thief.db') + thief.init_database(db_file) + monkeypatch.setattr('sys.argv', + ['thief', '-b', '1', '-H', 'cucm-empty', '--db', db_file]) + with pytest.raises(SystemExit): + thief.main() + out = capsys.readouterr().out + assert 'no MAC prefixes for cucm-empty' in out + assert 'You must specify at least one phone' not in out + + +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}) + monkeypatch.setattr(thief, 'get_servers_api', lambda *a, **kw: []) + db_file = str(tmp_path / 'thief.db') + thief.init_database(db_file) + monkeypatch.setattr('sys.argv', + ['thief', '--servers', '-H', 'cucm1', '--db', db_file]) + with pytest.raises(SystemExit): + thief.main() + assert len(calls) == 1 From 337d179e58f3070b23f78e487f7f86461681f91f Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Wed, 29 Jul 2026 13:45:02 -0400 Subject: [PATCH 2/4] feat: harden UDS probe for Contact Search Authentication (9443 + auth flag) Contact Search Authentication (CLI `utils contactsearchauthentication enable`) moves UDS off 8443 to 9443 and makes the /users resource require Basic auth. Both are cluster config, not tied to the CUCM major version (identical behaviour 11.5 through 15), so this is handled at runtime rather than version-gated. - get_version now parses and records the port it answered on. - New probe_uds() tries the requested port, then falls back to the other standard UDS port (8443<->9443) unless the user pinned --uds-port. The resolved port is threaded through every UDS feature call. - main() warns when the cluster requires auth for /users, so an empty --userenum/--directory/--spray reads as "auth required" instead of "no users", and notes when UDS was found on the alternate port. Confirmed against a live CUCM 14.0.1 target that the version endpoint is reachable unauthenticated on 8443; the fallback/auth paths are unit-tested. Co-Authored-By: Claude Fable 5 --- src/seeyoucm_thief/thief.py | 72 +++++++++++++++++---- tests/test_uds_hardening.py | 123 ++++++++++++++++++++++++++++++++++++ 2 files changed, 182 insertions(+), 13 deletions(-) create mode 100644 tests/test_uds_hardening.py diff --git a/src/seeyoucm_thief/thief.py b/src/seeyoucm_thief/thief.py index 0b05e66..4427c23 100644 --- a/src/seeyoucm_thief/thief.py +++ b/src/seeyoucm_thief/thief.py @@ -29,8 +29,14 @@ # Protocol ports # TFTP port is standard (69), HTTP_TFTP_PORT is configurable for fallback HTTP_TFTP_PORT = 6970 -# CUCM User Data Services (UDS) API — HTTPS only, default 8443 +# CUCM User Data Services (UDS) API — HTTPS only, default 8443. +# When Contact Search Authentication is enabled on the cluster (CLI +# `utils contactsearchauthentication enable`), UDS moves to 9443 and the +# /users resource then requires Basic auth. Both are version-independent +# (behaviour is identical on 11.5 through 15); the port is not gated on +# the CUCM major version. UDS_PORT = 8443 +UDS_PORT_SECURE = 9443 # Default output file for the standalone --directory harvest DEFAULT_DIRECTORY_OUTFILE = 'cucm_directory.csv' # Well-known default filenames the CUCM TFTP service hosts in addition to @@ -380,7 +386,8 @@ 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)'} + return {'version': '12.5.1-TEST', 'prefix': '11.0(1)', + 'usersAuthRequired': False, 'port': port} url = f'https://{cucm_host}:{port}/cucm-uds/version' dbg(f'UDS GET {url} (timeout={timeout}s)') @@ -399,9 +406,35 @@ def get_version(cucm_host, port=UDS_PORT, timeout=10): m = re.search(rf'<{field}>([^<]+)', resp.text) if m: info[field] = 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 + # it rather than letting the operator read that as "no users". + auth_m = re.search(r'\s*(true|false)\s*', + resp.text, re.IGNORECASE) + if auth_m: + info['usersAuthRequired'] = auth_m.group(1).lower() == 'true' + if info: + info['port'] = port return info or None +def probe_uds(cucm_host, port=UDS_PORT, allow_fallback=True, timeout=10): + """Locate a live UDS endpoint, tolerating the 8443<->9443 split that + Contact Search Authentication introduces. Tries ``port`` first; if that + yields nothing and ``allow_fallback`` is set, tries the other standard UDS + port. Returns the version-info dict (with a resolved ``port`` key) or None. + ``allow_fallback`` should be False when the user pinned --uds-port.""" + info = get_version(cucm_host, port=port, timeout=timeout) + if info or not allow_fallback: + return info + alt = UDS_PORT_SECURE if port == UDS_PORT else UDS_PORT + if alt == port: + return info + dbg(f'UDS version probe on :{port} failed; trying alternate port :{alt}') + return get_version(cucm_host, port=alt, timeout=timeout) + + def get_hostname_from_phone(phone_ip): if _TEST_MODE: return os.getenv("THIEF_TEST_PHONE_HOSTNAME") or "SEPTEST00000000" @@ -2841,21 +2874,34 @@ def main(): # timeout and print a misleading "could not retrieve version" error against # hosts where UDS is firewalled or not listening. uds_feature = args.servers or args.directory or args.userenum or args.spray + # Effective UDS port for the feature calls below. The probe may resolve it + # to 9443 when Contact Search Authentication has moved UDS off 8443; if the + # user pinned --uds-port we honour it and skip the fallback. + uds_port = args.uds_port if CUCM_host and uds_feature: - version_info = get_version(CUCM_host, port=args.uds_port) + allow_fallback = args.uds_port == UDS_PORT + version_info = probe_uds(CUCM_host, port=args.uds_port, allow_fallback=allow_fallback) 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 '')) + 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('usersAuthRequired'): + print('[!] This cluster requires authentication for the UDS /users resource ' + '(usersResourceAuthEnabled=true).') + print(' Unauthenticated --userenum/--directory/--spray will return no users; ' + 'valid CUCM credentials are needed.') else: - print(f'[-] Could not retrieve CUCM version from https://{CUCM_host}:{args.uds_port}/cucm-uds/version (run with -d for details)') + print(f'[-] Could not retrieve CUCM version from https://{CUCM_host}:{uds_port}/cucm-uds/version (run with -d for details)') if args.servers: if not CUCM_host: print('--servers requires -H/--host to specify the CUCM server') quit(1) - print(f'Enumerating CUCM cluster via https://{CUCM_host}:{args.uds_port}/cucm-uds/servers') - servers = get_servers_api(CUCM_host, port=args.uds_port) + print(f'Enumerating CUCM cluster via https://{CUCM_host}:{uds_port}/cucm-uds/servers') + servers = get_servers_api(CUCM_host, port=uds_port) if not servers: print('[-] No servers returned. Re-run with -d for request/response details.') quit(0) @@ -2882,8 +2928,8 @@ def main(): if not CUCM_host: print('--directory requires -H/--host to specify the CUCM server') quit(1) - print(f'Harvesting UDS directory from https://{CUCM_host}:{args.uds_port}/cucm-uds/users') - records = get_user_directory_api(CUCM_host, port=args.uds_port) + print(f'Harvesting UDS directory from https://{CUCM_host}:{uds_port}/cucm-uds/users') + records = get_user_directory_api(CUCM_host, port=uds_port) if not records: print('[-] No directory records returned. Re-run with -d for request/response details.') quit(0) @@ -2900,8 +2946,8 @@ def main(): if not CUCM_host: print('--userenum requires -H/--host to specify the CUCM server') quit(1) - print(f'Getting users from UDS API at https://{CUCM_host}:{args.uds_port}/cucm-uds/users') - api_users = get_users_api(CUCM_host, port=args.uds_port) + print(f'Getting users from UDS API at https://{CUCM_host}:{uds_port}/cucm-uds/users') + api_users = get_users_api(CUCM_host, port=uds_port) if api_users: unique_users = list(set(api_users)) with open(outfile, mode='w') as outputfile: @@ -2917,7 +2963,7 @@ def main(): if debug: for username in unique_users: print(f'{username}') - directory = get_user_directory_api(CUCM_host, port=args.uds_port) + directory = get_user_directory_api(CUCM_host, port=uds_port) if directory: if not no_db: written = record_uds_directory(CUCM_host, directory, db_file) @@ -2932,7 +2978,7 @@ def main(): if not no_db: print(f'[*] Probing UDS for associated devices (unauthenticated)...') found = enumerate_devices_unauthenticated( - CUCM_host, args.uds_port, unique_users, db_file, threads=threads, + CUCM_host, uds_port, unique_users, db_file, threads=threads, ) if found: print(f'[+] Found devices for {found} user(s) — attempting config downloads...') @@ -2985,7 +3031,7 @@ def main(): run_spray( cucm_host=CUCM_host, - port=args.uds_port, + port=uds_port, passwords=passwords, threads=args.spray_threads, rate_limit_hours=args.spray_rate_limit_hours, diff --git a/tests/test_uds_hardening.py b/tests/test_uds_hardening.py new file mode 100644 index 0000000..0ed381f --- /dev/null +++ b/tests/test_uds_hardening.py @@ -0,0 +1,123 @@ +"""Tests for UDS endpoint hardening: usersResourceAuthEnabled parsing and the +8443<->9443 fallback introduced by Contact Search Authentication.""" +from unittest.mock import MagicMock + +import pytest + +from seeyoucm_thief import thief + + +@pytest.fixture(autouse=True) +def _disable_test_mode(monkeypatch): + monkeypatch.setattr(thief, '_TEST_MODE', False) + + +def _version_xml(auth=None): + body = "14.0.114.0(1)" + if auth is not None: + body += f"{'true' if auth else 'false'}" + return body + "" + + +def _resp(text, status=200): + r = MagicMock() + r.status_code = status + r.text = text + r.content = text.encode() + return r + + +# --- get_version parsing --------------------------------------------------- + +def test_get_version_parses_auth_flag_true(monkeypatch): + monkeypatch.setattr(thief.requests, 'get', lambda *a, **kw: _resp(_version_xml(auth=True))) + info = thief.get_version('cucm', port=8443) + assert info['version'] == '14.0.1' + assert info['usersAuthRequired'] is True + assert info['port'] == 8443 + + +def test_get_version_parses_auth_flag_false(monkeypatch): + monkeypatch.setattr(thief.requests, 'get', lambda *a, **kw: _resp(_version_xml(auth=False))) + info = thief.get_version('cucm', port=8443) + assert info['usersAuthRequired'] is False + + +def test_get_version_absent_auth_flag_omitted(monkeypatch): + monkeypatch.setattr(thief.requests, 'get', lambda *a, **kw: _resp(_version_xml(auth=None))) + info = thief.get_version('cucm', port=9443) + assert 'usersAuthRequired' not in info + assert info['port'] == 9443 + + +# --- probe_uds fallback ---------------------------------------------------- + +def test_probe_uds_falls_back_to_9443(monkeypatch): + seen = [] + + def fake_get(url, **kw): + seen.append(url) + if ':8443/' in url: + raise thief.requests.exceptions.Timeout('read timed out') + return _resp(_version_xml(auth=True)) + + monkeypatch.setattr(thief.requests, 'get', fake_get) + info = thief.probe_uds('cucm', port=8443, allow_fallback=True) + assert info is not None + assert info['port'] == 9443 + assert any(':8443/' in u for u in seen) and any(':9443/' in u for u in seen) + + +def test_probe_uds_no_fallback_when_pinned(monkeypatch): + seen = [] + + def fake_get(url, **kw): + seen.append(url) + raise thief.requests.exceptions.Timeout('read timed out') + + monkeypatch.setattr(thief.requests, 'get', fake_get) + info = thief.probe_uds('cucm', port=8443, allow_fallback=False) + assert info is None + assert all(':9443/' not in u for u in seen) + + +def test_probe_uds_no_second_call_when_first_succeeds(monkeypatch): + seen = [] + + def fake_get(url, **kw): + seen.append(url) + return _resp(_version_xml(auth=False)) + + monkeypatch.setattr(thief.requests, 'get', fake_get) + info = thief.probe_uds('cucm', port=8443, allow_fallback=True) + assert info['port'] == 8443 + assert len(seen) == 1 + + +# --- main() wiring --------------------------------------------------------- + +def test_main_warns_when_auth_required(monkeypatch, tmp_path, capsys): + monkeypatch.setattr(thief, 'probe_uds', + lambda *a, **kw: {'version': '14.0.1', 'usersAuthRequired': True, 'port': 8443}) + 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 'usersResourceAuthEnabled=true' in out + + +def test_main_threads_resolved_port_to_feature(monkeypatch, tmp_path): + monkeypatch.setattr(thief, 'probe_uds', + lambda *a, **kw: {'version': '14.0.1', 'port': 9443}) + used = {} + monkeypatch.setattr(thief, 'get_servers_api', + lambda host, port=8443, **kw: used.update(port=port) or []) + db = tmp_path / 'thief.db' + thief.init_database(str(db)) + monkeypatch.setattr('sys.argv', ['thief', '--servers', '-H', 'cucm', '--db', str(db)]) + with pytest.raises(SystemExit): + thief.main() + assert used['port'] == 9443 From 545dcf787b4bd6a07fbc14a0732af9e9cb5b487c Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Wed, 29 Jul 2026 13:47:28 -0400 Subject: [PATCH 3/4] chore(ci): run PyTest on dev branch pushes and PRs dev is now an integration branch; without this it merged untested. Co-Authored-By: Claude Fable 5 --- .github/workflows/pytest.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/pytest.yaml b/.github/workflows/pytest.yaml index 2e39eee..b19e248 100644 --- a/.github/workflows/pytest.yaml +++ b/.github/workflows/pytest.yaml @@ -2,9 +2,9 @@ name: PyTest on: push: - branches: [ main ] + branches: [ main, dev ] pull_request: - branches: [ main ] + branches: [ main, dev ] jobs: test: runs-on: ubuntu-latest From df46d3d3e0e7f6e7dcc8fbdb2f11d2cc155fdb16 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Wed, 29 Jul 2026 14:05:19 -0400 Subject: [PATCH 4/4] fix: recognize SEP names in -p (and resolve their CUCM from the DB) A -p value that already names the device (SEPC064E4D83AAF, optionally with a DNS suffix like SEPC064E4D83AAF.mason.ad) carries its 12-hex MAC directly, but --brute-mac always tried to HTTP-scrape the MAC off the phone's NetworkConfiguration page. An unreachable phone therefore reported "could not detect MAC" even though the MAC was in the argument. - New mac_from_phone_arg(): extracts the MAC from a SEP value; the detect worker uses it and skips the HTTP lookup for such values. - New get_cucm_for_mac_from_db(): when no -H is given and the phone is unreachable, resolve the device's CUCM from mac_prefixes/uds_devices recorded on an earlier scan, so a SEP name already known in the database just works. Otherwise the error now tells the operator to supply -H. Co-Authored-By: Claude Fable 5 --- src/seeyoucm_thief/thief.py | 64 +++++++++++++++++++++++- tests/test_brute_host_prefixes.py | 83 +++++++++++++++++++++++++++++++ 2 files changed, 145 insertions(+), 2 deletions(-) diff --git a/src/seeyoucm_thief/thief.py b/src/seeyoucm_thief/thief.py index 4427c23..f4704ae 100644 --- a/src/seeyoucm_thief/thief.py +++ b/src/seeyoucm_thief/thief.py @@ -435,6 +435,18 @@ def probe_uds(cucm_host, port=UDS_PORT, allow_fallback=True, timeout=10): return get_version(cucm_host, port=alt, timeout=timeout) +def mac_from_phone_arg(phone): + """If a -p value already names the device (a SEP hostname, optionally + with a DNS suffix like SEPC064E4D83AAF.mason.ad), the 12-hex MAC is in the + argument itself. Return it uppercased, or None if the value is a plain + IP/hostname that must be contacted to learn its MAC. Anchored to SEP so a + stray 12-hex run elsewhere in a hostname is not misread as a MAC.""" + if not phone: + return None + m = re.search(r'SEP([0-9A-Fa-f]{12})(?![0-9A-Fa-f])', phone, re.IGNORECASE) + return m.group(1).upper() if m else None + + def get_hostname_from_phone(phone_ip): if _TEST_MODE: return os.getenv("THIEF_TEST_PHONE_HOSTNAME") or "SEPTEST00000000" @@ -1043,6 +1055,39 @@ def get_uds_device_macs_from_db(cucm_host, db_file='thief.db'): return rows +def get_cucm_for_mac_from_db(full_mac, db_file='thief.db'): + """Return the CUCM host previously associated with a given device MAC, or + None. Lets ``--brute-mac -p SEP`` resolve its CUCM from an earlier + scan when the phone itself is unreachable and no -H was supplied. Checks + mac_prefixes first (phone scans), then uds_devices (UDS harvests).""" + if not full_mac: + return None + mac = full_mac.upper() + try: + conn = sqlite3.connect(db_file, timeout=30.0) + try: + for sql, param in ( + ('SELECT cucm_host FROM mac_prefixes WHERE UPPER(full_mac) = ? ' + 'AND cucm_host IS NOT NULL ORDER BY discovery_time DESC LIMIT 1', mac), + ('SELECT cucm_host FROM uds_devices WHERE UPPER(device_name) = ? ' + 'AND cucm_host IS NOT NULL ORDER BY discovery_time DESC LIMIT 1', f'SEP{mac}'), + ): + try: + row = conn.execute(sql, (param,)).fetchone() + except sqlite3.OperationalError as e: + if 'no such table' not in str(e): + raise + continue + if row and row[0]: + return row[0] + finally: + conn.close() + except Exception as e: + if globals().get('debug', False): + print(f'[!] get_cucm_for_mac_from_db error: {e}') + return None + + def parse_uds_devices(xml_body): """Extract SEP device names from a /cucm-uds/user/{id} XML response body.""" return re.findall(r'(SEP[0-9A-Fa-f]{12})', xml_body) @@ -3111,8 +3156,15 @@ def detect_worker(): break try: index = phone_index.get(phone, 0) + 1 - _safe_print(f'[{index}/{len(phones)}] Detecting MAC address from phone {phone}...') - hostname = get_hostname_from_phone(phone) + # A SEP value carries the MAC directly; only fall back + # to an HTTP lookup for plain IPs/hostnames, so an + # unreachable phone named by its SEP address still works. + if mac_from_phone_arg(phone): + _safe_print(f'[{index}/{len(phones)}] Using MAC from device name {phone}...') + hostname = phone + else: + _safe_print(f'[{index}/{len(phones)}] Detecting MAC address from phone {phone}...') + hostname = get_hostname_from_phone(phone) if hostname: mac_match = re.search(r'SEP([0-9A-F]{12})', hostname, re.IGNORECASE) if mac_match: @@ -3131,8 +3183,16 @@ def detect_worker(): phone_cucm = CUCM_host else: phone_cucm = get_cucm_name_from_phone(phone) + if not phone_cucm and not no_db: + # SEP name given but phone unreachable: the + # device may already be tied to a CUCM in the + # database from an earlier scan. + phone_cucm = get_cucm_for_mac_from_db(full_mac, db_file) + if phone_cucm: + _safe_print(f' ✓ CUCM {phone_cucm} resolved from database for SEP{full_mac}') if not phone_cucm: _safe_print(f' ✗ Could not detect CUCM host from phone {phone}') + _safe_print(f' → Supply -H (the phone was not reachable to auto-detect it)') _safe_print(f' → Skipping this phone, continuing with others...\n') with detect_lock: counts["fail"] += 1 diff --git a/tests/test_brute_host_prefixes.py b/tests/test_brute_host_prefixes.py index c01ad3d..0128907 100644 --- a/tests/test_brute_host_prefixes.py +++ b/tests/test_brute_host_prefixes.py @@ -97,3 +97,86 @@ def test_servers_feature_still_probes_uds_version(monkeypatch, tmp_path): with pytest.raises(SystemExit): thief.main() assert len(calls) == 1 + + +def test_mac_from_phone_arg_plain_sep(): + assert thief.mac_from_phone_arg('SEPC064E4D83AAF') == 'C064E4D83AAF' + + +def test_mac_from_phone_arg_with_dns_suffix(): + assert thief.mac_from_phone_arg('SEPC064E4D83AAF.mason.ad') == 'C064E4D83AAF' + + +def test_mac_from_phone_arg_lowercase_normalised(): + assert thief.mac_from_phone_arg('sepc064e4d83aaf') == 'C064E4D83AAF' + + +def test_mac_from_phone_arg_ip_returns_none(): + assert thief.mac_from_phone_arg('10.45.200.50') is None + + +def test_mac_from_phone_arg_plain_hostname_returns_none(): + assert thief.mac_from_phone_arg('phone-lobby.example.com') is None + + +def test_mac_from_phone_arg_empty(): + assert thief.mac_from_phone_arg('') is None + + +def test_brute_mac_sep_name_skips_http_detection(monkeypatch, tmp_path, capsys): + """A SEP passed to -p must not trigger an HTTP phone lookup and must + still yield its MAC when the phone is unreachable.""" + def _boom(*a, **kw): + raise AssertionError('get_hostname_from_phone should not be called for a SEP name') + monkeypatch.setattr(thief, 'get_hostname_from_phone', _boom) + monkeypatch.setattr(thief, 'get_version', lambda *a, **kw: None) + db_file = str(tmp_path / 'thief.db') + thief.init_database(db_file) + monkeypatch.setattr('sys.argv', + ['thief', '-b', '3', '-p', 'SEPC064E4D83AAF.mason.ad', + '-H', 'cucm1', '--db', db_file]) + with pytest.raises(SystemExit): + thief.main() + out = capsys.readouterr().out + assert 'Using MAC from device name' in out + assert 'Detected: SEPC064E4D83AAF' in out + assert 'Could not detect hostname' not in out + + +def test_get_cucm_for_mac_from_db_mac_prefixes(tmp_path): + db_file = _db(tmp_path) + thief.log_mac_prefix_to_db('cucm-a', '10.0.0.5', 'C064E4D83AAF', 'C064E4D83', db_file) + assert thief.get_cucm_for_mac_from_db('c064e4d83aaf', db_file) == 'cucm-a' + assert thief.get_cucm_for_mac_from_db('AABBCCDDEEFF', db_file) is None + + +def test_get_cucm_for_mac_from_db_uds_devices(tmp_path): + db_file = _db(tmp_path) + thief.log_uds_device('cucm-b', 'alice', 'SEPC064E4D83AAF', 'userenum', db_file) + assert thief.get_cucm_for_mac_from_db('C064E4D83AAF', db_file) == 'cucm-b' + + +def test_get_cucm_for_mac_from_db_missing_tables(tmp_path): + import sqlite3 as _sq + db_file = str(tmp_path / 'empty.db') + _sq.connect(db_file).close() + assert thief.get_cucm_for_mac_from_db('C064E4D83AAF', db_file) is None + + +def test_brute_mac_sep_name_resolves_cucm_from_db(monkeypatch, tmp_path, capsys): + """-p SEP with no -H and an unreachable phone should still find its + CUCM from a prior scan in the database.""" + monkeypatch.setattr(thief, 'get_hostname_from_phone', + lambda *a, **kw: (_ for _ in ()).throw(AssertionError('no http'))) + monkeypatch.setattr(thief, 'get_cucm_name_from_phone', lambda *a, **kw: None) + monkeypatch.setattr(thief, 'get_version', lambda *a, **kw: None) + db_file = str(tmp_path / 'thief.db') + thief.init_database(db_file) + thief.log_uds_device('cucm-b', 'alice', 'SEPC064E4D83AAF', 'userenum', db_file) + monkeypatch.setattr('sys.argv', + ['thief', '-b', '3', '-p', 'SEPC064E4D83AAF.mason.ad', '--db', db_file]) + with pytest.raises(SystemExit): + thief.main() + out = capsys.readouterr().out + assert 'resolved from database' in out + assert 'cucm-b' in out