"""Remove a deleted VM's entries from local OpenSSH host-key caches.""" from __future__ import annotations import base64 import binascii import fcntl import hmac import os import pwd import re import stat import uuid from ipaddress import IPv4Address from pathlib import Path from .config import Settings from .domain import VmRecord from .errors import UvmError _HOST_LINE = re.compile( rb"^[ \t]*(?:@(?:cert-authority|revoked)[ \t]+)?" rb"(?P[^ \t\r\n]+)[ \t]+[^ \t\r\n]+[ \t]+[^ \t\r\n]+" ) _PORT_HOST = re.compile(rb"\[([^\]]+)\]:(\d+)\Z") class KnownHostsManager: """Deletion-only cleanup; lifecycle owns the lock and confirms VM teardown.""" def __init__(self, settings: Settings) -> None: self._settings = settings def remove(self, vm: VmRecord) -> None: try: address = str(IPv4Address(vm.guest_ip)).encode("ascii") except ValueError as error: raise UvmError("known_hosts cleanup requires a valid guest IPv4 address") from error if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,80}", vm.id) is None: raise UvmError("known_hosts cleanup requires a valid VM ID") alias = f"uvm-{vm.id}".encode("ascii") hosts = {address, alias, alias.lower()} ports = {22, *(rule.port for rule in vm.ports or ())} hashed_candidates = hosts | { b"[" + host + b"]:" + str(port).encode("ascii") for host in hosts for port in ports } for path, owner in self._files().items(): self._remove_file(path, hosts, hashed_candidates, owner) def _files(self) -> dict[Path, int | None]: if self._settings.known_hosts_files is not None: return dict.fromkeys(self._settings.known_hosts_files) try: effective_uid = os.geteuid() accounts = [pwd.getpwuid(effective_uid)] if effective_uid == 0: sudo_uid = os.environ.get("SUDO_UID") sudo_user = os.environ.get("SUDO_USER") if sudo_uid: if not sudo_uid.isascii() or not sudo_uid.isdigit(): raise ValueError("invalid SUDO_UID") accounts.append(pwd.getpwuid(int(sudo_uid))) elif sudo_user: accounts.append(pwd.getpwnam(sudo_user)) return { Path(account.pw_dir) / ".ssh" / filename: account.pw_uid for account in accounts for filename in ("known_hosts", "known_hosts2") } except (KeyError, ValueError, OverflowError) as error: raise UvmError("could not determine SSH user homes; configure UVM_KNOWN_HOSTS_FILES") from error @staticmethod def _remove_file(path: Path, hosts: set[bytes], candidates: set[bytes], owner: int | None) -> None: directory = None temporary = None try: # Pin the directory and reject symlink/FIFO/hard-link targets when # editing another local user's cache from a privileged process. try: directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) descriptor = os.open(path.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory) except FileNotFoundError: return with os.fdopen(descriptor, "rb") as source: metadata = os.fstat(source.fileno()) if not stat.S_ISREG(metadata.st_mode): raise UvmError(f"known_hosts must be a regular, single-link file: {path}") fcntl.flock(source.fileno(), fcntl.LOCK_EX) metadata = os.fstat(source.fileno()) if metadata.st_nlink != 1: raise UvmError(f"known_hosts must be a regular, single-link file: {path}") if owner is not None and metadata.st_uid != owner: raise UvmError(f"known_hosts has an unexpected owner: {path}") contents = source.read() updated = _without_hosts(contents, hosts, candidates) if updated == contents: return candidate = f".{path.name}.uvm-{uuid.uuid4().hex}" descriptor = os.open(candidate, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600, dir_fd=directory) temporary = candidate with os.fdopen(descriptor, "wb") as output: output.write(updated) output.flush() created = os.fstat(output.fileno()) if (created.st_uid, created.st_gid) != (metadata.st_uid, metadata.st_gid): os.fchown(output.fileno(), metadata.st_uid, metadata.st_gid) os.fchmod(output.fileno(), stat.S_IMODE(metadata.st_mode)) for attribute in os.listxattr(source.fileno()): os.setxattr(output.fileno(), attribute, os.getxattr(source.fileno(), attribute)) os.fsync(output.fileno()) current = os.stat(path.name, dir_fd=directory, follow_symlinks=False) fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns", "st_mode", "st_uid", "st_gid", "st_nlink") if any(getattr(current, field) != getattr(metadata, field) for field in fields): raise UvmError(f"known_hosts changed during cleanup; retry deletion: {path}") os.replace(temporary, path.name, src_dir_fd=directory, dst_dir_fd=directory) os.fsync(directory) except OSError as error: raise UvmError(f"could not clean SSH known_hosts {path}: {error}") from error finally: if directory is not None: try: if temporary is not None: try: os.unlink(temporary, dir_fd=directory) except FileNotFoundError: pass finally: os.close(directory) def _without_hosts(contents: bytes, hosts: set[bytes], candidates: set[bytes]) -> bytes: """Preserve unrelated aliases, comments, markers, key types, and line endings.""" def matches(token: bytes) -> bool: if token.startswith(b"|1|"): try: _, _version, salt_text, digest_text = token.split(b"|") salt = base64.b64decode(salt_text, validate=True) digest = base64.b64decode(digest_text, validate=True) except (ValueError, binascii.Error): return False return len(salt) == len(digest) == 20 and any( hmac.compare_digest(digest, hmac.digest(salt, candidate, "sha1")) for candidate in candidates ) token = token.lower() if token in hosts: return True bracketed = _PORT_HOST.fullmatch(token) return (bracketed is not None and bracketed[1] in hosts and len(bracketed[2]) <= 5 and 1 <= int(bracketed[2]) <= 65535) retained = [] for line in contents.splitlines(keepends=True): if line.lstrip().startswith(b"#"): retained.append(line) continue match = _HOST_LINE.match(line) if match is None: retained.append(line) continue names = match["hosts"].split(b",") remaining = [name for name in names if not matches(name)] if remaining == names: retained.append(line) elif remaining: retained.append(line[:match.start("hosts")] + b",".join(remaining) + line[match.end("hosts"):]) return b"".join(retained)