import boto3 import json import os import subprocess import base64 from datetime import datetime from botocore.exceptions import ClientError import openpyxl from openpyxl.utils import get_column_letter region = os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION") client = boto3.client("secretsmanager", region_name=region) if region else boto3.client("secretsmanager") ENV = "prod" STORE_IDS = { "1", "3", "11", "12", "117", "167", "173", "187", "206", "213", "286", "309", "337", "339", "348", "365", "384", "399", "404", "412", "428", "446", "452", "453", "461", "465", "493", "495", "513", "523", "526", "534", "535", "538", "545", "549", "552", "553", "565", "568", "569", "570", "572", "575", "581", "616", "644", "670", "677", "694", "695", "696", "698", "705", "706", "707", "708", "711", "723", "737", "738", "739", "740", "752", "836", "838", "854", "1101", "1105", "1129", "1135", "1173", "1208", "1212", "1213", "1223", "1245", "1261", "1262", "1277", "1278", "1279", "1294", "1308", "1309", "1331", "1350", "1351", "1371", "1377", "1378", "1389", "1395", "1414", "1433", "1438", "1448", "1450", "1454", "1459", "1469", "1470", "1479", "1484", "1498", "1504", "1505", "1506", "1507", "1515", "1517", "1524", "1532", "1534", "1545", "1548", "1551", "1584", "1591", "1592", "1637", "1668", "1685", "1686", "1692", "1697", "1700", "1702", "1704", "1710", "1711", "1727", "1775", "1809", "1810", "1823", "1829", "1830", "1831", # "1838", # skipping this store id since it has hardcoded host_key_algorithm "1842", "1850", "1861", "1866", "1876", "1879", "1892", "1893", "1895", "1896", "1907", "1927", "1928", "1942", "1950", "1967", } KEYSCAN_TIMEOUT = "5" KEYSCAN_ALGOS = "ed25519,rsa" KNOWN_ALGO_NAMES = [ "ssh-ed25519", "ssh-rsa" ] ALGO_PRIORITY = [ "ssh-ed25519", "ssh-rsa" ] #In case of mismatch between known and offered algorithms, recommend the best offered algo based on our priority list, even if it's not the known one. def best_offered(offered): for algo in ALGO_PRIORITY: if algo in offered: return algo return next(iter(offered)) def decode_known_host(raw): if not raw or not raw.strip(): return "" raw = raw.strip() for algo in KNOWN_ALGO_NAMES: if raw.startswith(algo): return algo try: decoded = base64.b64decode(raw) decoded_str = decoded.decode("utf-8", errors="ignore") for algo in KNOWN_ALGO_NAMES: if decoded_str.startswith(algo): return algo for algo in KNOWN_ALGO_NAMES: if algo.encode("ascii") in decoded: return algo except Exception: pass return "" def discover_offered_algos(domain, port): algos = set() for algo_type in KEYSCAN_ALGOS.split(","): try: result = subprocess.run( ["ssh-keyscan", "-T", KEYSCAN_TIMEOUT, "-p", port, "-t", algo_type, domain], text=True, capture_output=True, timeout=10, ) except subprocess.TimeoutExpired: continue for line in result.stdout.splitlines(): parts = line.split() if len(parts) >= 2 and not line.startswith("#"): algos.add(parts[1]) return algos def get_store_id(name): parts = name.split("/") for i, part in enumerate(parts): if part == "connection_info" and i + 1 < len(parts): return parts[i + 1] return parts[0] def get_order_type(name): parts = name.split("/") for i, part in enumerate(parts): if part == "connection_info" and i + 2 < len(parts): return parts[i + 2] return "" HEADERS = [ "Store ID", "Order type", "Connection type", "Domain name", "Port", "Private key file", "Algorithms offered by the server", "Recommended algo", "Status", ] def write_excel(rows, output_path): wb = openpyxl.Workbook() ws = wb.active ws.title = f"SFTP Audit {ENV.upper()}" ws.append(HEADERS) for row in rows: ws.append(row) col_widths = [12, 12, 16, 40, 8, 20, 45, 45, 32] for col_idx, width in enumerate(col_widths, start=1): ws.column_dimensions[get_column_letter(col_idx)].width = width wb.save(output_path) print(f"\nExcel saved to: {output_path}") def process_secrets(): prefix = f"{ENV}/direct_delivery/connection_info" count = 0 rows = [] paginator = client.get_paginator("list_secrets") for page in paginator.paginate(Filters=[{"Key": "name", "Values": [prefix]}]): for secret in page["SecretList"]: name = secret.get("Name", "") if prefix not in name: continue store_id = get_store_id(name) order_type = get_order_type(name) if store_id not in STORE_IDS: continue try: secret_response = client.get_secret_value(SecretId=name) parsed = json.loads(secret_response["SecretString"]) conn_type = parsed.get("connection_type", "").lower() if conn_type != "sftp": continue domain = parsed.get("domain_name", "") port = str(parsed.get("port", "22")) known_host = parsed.get("known_host", "") private_key_file = parsed.get("private_key_file", parsed.get("key_file", "")) known_algo = decode_known_host(known_host) print(f"[Scanning {count + 1}] {name} ({domain}:{port})...", flush=True) offered = discover_offered_algos(domain, port) if not offered: status = "unreachable_or_no_ssh_key" recommended = known_algo or "" elif known_algo and known_algo not in offered: status = "known_algo_not_offered_by_host" recommended = best_offered(offered) else: status = "ok" recommended = known_algo if known_algo else best_offered(offered) offered_display = "|".join(sorted(offered)) if offered else "unable_to_fetch" print(f"Secret: {name}") print(f" Domain: {domain}:{port}") print(f" Known host algo: {known_algo or '(none)'}") print(f" Offered algos: {offered_display}") print(f" Recommended: {recommended}") print(f" Status: {status}") print("-" * 50) rows.append([store_id, order_type, conn_type, domain, port, private_key_file, offered_display, recommended, status]) count += 1 except ClientError as e: if e.response["Error"]["Code"] == "AccessDeniedException": print(f"Skipping {name}: Access denied") else: print(f"Error fetching {name}: {e}") print(f"\nTotal processed: {count}") if rows: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_path = f"sftp_audit_{ENV}_{timestamp}.xlsx" write_excel(rows, output_path) if __name__ == "__main__": process_secrets()