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 diff --git a/src/seeyoucm_thief/thief.py b/src/seeyoucm_thief/thief.py index 0b05e66..f4704ae 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,47 @@ 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 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" @@ -1010,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) @@ -2841,21 +2919,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 +2973,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 +2991,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 +3008,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 +3023,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 +3076,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, @@ -3065,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: @@ -3085,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 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