admin/trailofbits-dropkit
1
0
Fork 0
mirror of https://github.com/trailofbits/dropkit synced 2026-06-21 14:11:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix known_hosts removal for hashed entries (#28)

Browse Source 
Replace manual file parsing with ssh-keygen -R which properly handles
both hashed (|1|...) and unhashed entries. macOS and many Linux systems
use HashKnownHosts by default, making the previous implementation unable
to remove entries.

Changes:
- Use ssh-keygen -R for each hostname instead of parsing the file
- Detect successful removal via "updated" in stdout
- Backup files are now .old (ssh-keygen default) instead of .bak
- Bracketed entries ([host]:port) now require exact format

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Riccardo Schirone
2026-01-30 21:58:05 +01:00
committed by GitHub
parent a875f3d430
commit 9895dd9d5e
3 changed files with 226 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dropkit/main.py
+128 -10
View File
@@ -343,7 +343,11 @@ def ensure_ssh_config(
def cleanup_ssh_entries(
config: DropkitConfig, droplet_name: str, prompt_known_hosts: bool = True
config: DropkitConfig,
droplet_name: str,
prompt_known_hosts: bool = True,
public_ip: str | None = None,
tailscale_ip: str | None = None,
) -> None:
"""
Remove SSH config entry and optionally clean up known_hosts for a droplet.
@@ -352,6 +356,10 @@ def cleanup_ssh_entries(
config: DropkitConfig instance with SSH settings.
droplet_name: Name of the droplet being removed.
prompt_known_hosts: If True, prompt user before removing known_hosts entries.
public_ip: Public IP of the droplet (for known_hosts cleanup when SSH config
has a different IP, e.g., Tailscale IP).
tailscale_ip: Tailscale IP of the droplet (for known_hosts cleanup when SSH
config has the public IP but user also connected via Tailscale).
"""
ssh_hostname = get_ssh_hostname(droplet_name)
@@ -381,6 +389,12 @@ def cleanup_ssh_entries(
hostnames_to_remove = [ssh_hostname]
if ssh_ip:
hostnames_to_remove.append(ssh_ip)
# Include public IP if different from SSH config IP (e.g., when using Tailscale)
if public_ip and public_ip not in hostnames_to_remove:
hostnames_to_remove.append(public_ip)
# Include Tailscale IP if different from SSH config IP
if tailscale_ip and tailscale_ip not in hostnames_to_remove:
hostnames_to_remove.append(tailscale_ip)
console.print(f"[dim]Hostnames to find: {hostnames_to_remove}[/dim]")
try:
removed = remove_known_hosts_entry(known_hosts_path, hostnames_to_remove)
@@ -1041,6 +1055,51 @@ def prepare_for_hibernate(
return True
def get_tailscale_ip(ssh_hostname: str) -> str | None:
"""
Get Tailscale IP address from a running droplet.
SSHes to the droplet and runs 'tailscale ip -4' to get the Tailscale IP.
This is a single attempt (not polling) and will return None if unreachable
or Tailscale is not running.
Args:
ssh_hostname: SSH hostname to connect to
Returns:
Tailscale IP address if found, None otherwise
"""
try:
result = subprocess.run(
[
"ssh",
"-o",
"StrictHostKeyChecking=no",
"-o",
"UserKnownHostsFile=/dev/null",
"-o",
"ConnectTimeout=5",
"-o",
"BatchMode=yes",
ssh_hostname,
"tailscale ip -4 2>/dev/null",
],
capture_output=True,
timeout=15,
)
output = result.stdout.decode("utf-8", errors="ignore").strip()
# Validate it's a valid Tailscale CGNAT IP (100.64.0.0/10 range)
if output and is_tailscale_ip(output):
return output
except (subprocess.TimeoutExpired, subprocess.SubprocessError):
pass
return None
def wait_for_tailscale_ip(
ssh_hostname: str,
timeout: int = 300,
@@ -2560,11 +2619,13 @@ def destroy(droplet_name: str = typer.Argument(..., autocompletion=complete_drop
info_table.add_row("Status:", status)
# Get IP address
droplet_public_ip = None
networks = droplet.get("networks", {})
v4_networks = networks.get("v4", [])
for network in v4_networks:
if network.get("type") == "public":
info_table.add_row("IP:", network.get("ip_address", "N/A"))
droplet_public_ip = network.get("ip_address")
info_table.add_row("IP:", droplet_public_ip or "N/A")
break
info_table.add_row("Region:", droplet.get("region", {}).get("slug", "N/A"))
@@ -2606,9 +2667,15 @@ def destroy(droplet_name: str = typer.Argument(..., autocompletion=complete_drop
console.print("[dim]Cancelled.[/dim]")
raise typer.Exit(1)
# Logout from Tailscale if droplet is Tailscale-locked (best-effort)
if is_droplet_tailscale_locked(config, droplet_name):
ssh_hostname = get_ssh_hostname(droplet_name)
# Get Tailscale IP before destroying (for known_hosts cleanup)
ssh_hostname = get_ssh_hostname(droplet_name)
droplet_tailscale_ip = None
tailscale_locked = is_droplet_tailscale_locked(config, droplet_name)
if tailscale_locked:
# SSH config has the Tailscale IP - save it for cleanup
droplet_tailscale_ip = get_ssh_host_ip(config.ssh.config_path, ssh_hostname)
# Try to logout from Tailscale
console.print()
console.print("[dim]Logging out from Tailscale...[/dim]")
if tailscale_logout(ssh_hostname):
@@ -2618,6 +2685,12 @@ def destroy(droplet_name: str = typer.Argument(..., autocompletion=complete_drop
"[yellow]⚠[/yellow] Could not logout from Tailscale "
"(device may remain in Tailscale admin console)"
)
else:
# SSH config has public IP, try to get Tailscale IP via SSH
console.print("[dim]Checking for Tailscale IP...[/dim]")
droplet_tailscale_ip = get_tailscale_ip(ssh_hostname)
if droplet_tailscale_ip:
console.print(f"[dim]Found Tailscale IP: {droplet_tailscale_ip}[/dim]")
# Proceed with deletion
console.print()
@@ -2627,7 +2700,13 @@ def destroy(droplet_name: str = typer.Argument(..., autocompletion=complete_drop
console.print("[green]✓[/green] Droplet destroyed")
# Remove SSH config entry and clean up known_hosts
cleanup_ssh_entries(config, droplet_name, prompt_known_hosts=True)
cleanup_ssh_entries(
config,
droplet_name,
prompt_known_hosts=True,
public_ip=droplet_public_ip,
tailscale_ip=droplet_tailscale_ip,
)
console.print()
console.print("[bold green]Droplet successfully destroyed[/bold green]")
@@ -2646,6 +2725,8 @@ def _complete_hibernate(
username: str,
size_slug: str,
tailscale_locked: bool = False,
public_ip: str | None = None,
tailscale_ip: str | None = None,
) -> None:
"""
Complete hibernate after snapshot is done (tag, destroy, cleanup, success message).
@@ -2665,6 +2746,8 @@ def _complete_hibernate(
username: Username for tagging
size_slug: Size slug for tagging
tailscale_locked: If True, tag snapshot with tailscale-lockdown
public_ip: Public IP of the droplet (for known_hosts cleanup)
tailscale_ip: Tailscale IP of the droplet (for known_hosts cleanup)
"""
user_tag = get_user_tag(username)
@@ -2694,7 +2777,13 @@ def _complete_hibernate(
console.print("[green]✓[/green] Droplet destroyed")
# Remove SSH config entry and clean up known_hosts
cleanup_ssh_entries(config, droplet_name, prompt_known_hosts=True)
cleanup_ssh_entries(
config,
droplet_name,
prompt_known_hosts=True,
public_ip=public_ip,
tailscale_ip=tailscale_ip,
)
# Summary
console.print()
@@ -3399,6 +3488,32 @@ def hibernate(
status = droplet.get("status", "")
size_slug = droplet.get("size_slug", "")
# Extract public IP for known_hosts cleanup
droplet_public_ip = None
networks = droplet.get("networks", {})
v4_networks = networks.get("v4", [])
for network in v4_networks:
if network.get("type") == "public":
droplet_public_ip = network.get("ip_address")
break
# Get Tailscale IP for known_hosts cleanup
# Must be done BEFORE prepare_for_tailscale_hibernate() changes the SSH config
ssh_hostname = get_ssh_hostname(droplet_name)
droplet_tailscale_ip = None
tailscale_locked = is_droplet_tailscale_locked(config, droplet_name)
if tailscale_locked:
# SSH config currently has the Tailscale IP - save it before it gets changed
droplet_tailscale_ip = get_ssh_host_ip(config.ssh.config_path, ssh_hostname)
if droplet_tailscale_ip:
console.print(f"[dim]Saved Tailscale IP: {droplet_tailscale_ip}[/dim]")
else:
# SSH config has public IP, try to get Tailscale IP via SSH
console.print("[dim]Checking for Tailscale IP...[/dim]")
droplet_tailscale_ip = get_tailscale_ip(ssh_hostname)
if droplet_tailscale_ip:
console.print(f"[dim]Found Tailscale IP: {droplet_tailscale_ip}[/dim]")
# Generate snapshot name
snapshot_name = get_snapshot_name(droplet_name)
@@ -3438,12 +3553,11 @@ def hibernate(
elif action_status == "completed":
console.print("[green]✓[/green] Snapshot already completed")
# Detect Tailscale lockdown before completing
tailscale_locked = is_droplet_tailscale_locked(config, droplet_name)
# Complete the hibernate (tag, destroy, cleanup)
# Note: tailscale_locked was already detected earlier
if tailscale_locked:
console.print("[dim]Detected Tailscale lockdown[/dim]")
# Complete the hibernate (tag, destroy, cleanup)
_complete_hibernate(
api,
config,
@@ -3453,6 +3567,8 @@ def hibernate(
username,
size_slug,
tailscale_locked=tailscale_locked,
public_ip=droplet_public_ip,
tailscale_ip=droplet_tailscale_ip,
)
return # Exit early, skip normal flow
@@ -3560,6 +3676,8 @@ def hibernate(
username,
size_slug,
tailscale_locked=tailscale_locked,
public_ip=droplet_public_ip,
tailscale_ip=droplet_tailscale_ip,
)
except DigitalOceanAPIError as e:
dropkit/ssh_config.py
+16 -59
View File
@@ -1,6 +1,7 @@
"""SSH config file management."""
import shutil
import subprocess
from pathlib import Path
@@ -226,78 +227,34 @@ def remove_known_hosts_entry(known_hosts_path: str, hostnames: list[str]) -> int
"""
Remove entries for specified hostnames from known_hosts file.
Uses ssh-keygen -R which properly handles hashed entries.
Args:
known_hosts_path: Path to known_hosts file (e.g., ~/.ssh/known_hosts)
hostnames: List of hostnames/IPs to remove
Returns:
Number of entries removed
Number of hostnames for which entries were removed
"""
known_hosts_file = Path(known_hosts_path).expanduser()
if not known_hosts_file.exists():
return 0
# Backup existing known_hosts before modifying
_backup_config(known_hosts_file)
# Read existing known_hosts
with open(known_hosts_file) as f:
lines = f.readlines()
# Normalize hostnames for matching (lowercase)
hostnames_lower = {h.lower() for h in hostnames}
# Filter out matching entries
new_lines = []
removed_count = 0
for line in lines:
stripped = line.strip()
# Skip empty lines and comments, preserve them
if not stripped or stripped.startswith("#"):
new_lines.append(line)
continue
# Skip hashed entries (can't match them)
if stripped.startswith("|1|"):
new_lines.append(line)
continue
# Parse the first field (comma-separated hostnames)
# Format: hostname[,hostname2,...] keytype key [comment]
parts = stripped.split(None, 1)
if not parts:
new_lines.append(line)
continue
host_field = parts[0]
entry_hosts = host_field.split(",")
# Check if any of our hostnames match any entry host
should_remove = False
for entry_host in entry_hosts:
# Handle bracketed entries like [hostname]:port
check_host = entry_host.lower()
if check_host.startswith("["):
# Extract hostname from [hostname]:port
bracket_end = check_host.find("]")
if bracket_end > 0:
check_host = check_host[1:bracket_end]
if check_host in hostnames_lower:
should_remove = True
break
if should_remove:
for hostname in hostnames:
result = subprocess.run(
["ssh-keygen", "-R", hostname, "-f", str(known_hosts_file)],
capture_output=True,
text=True,
)
# ssh-keygen -R returns 0 even if hostname not found
# When it removes an entry, it prints to stdout:
# "# Host <hostname> found: line N"
# "<file> updated."
# "Original contents retained as <file>.old"
if "updated" in result.stdout:
removed_count += 1
else:
new_lines.append(line)
# Only write if we removed something
if removed_count > 0:
with open(known_hosts_file, "w") as f:
f.writelines(new_lines)
return removed_count
tests/test_ssh_config.py
+82 -16
View File
@@ -839,13 +839,18 @@ otherhost ssh-rsa AAAA...
assert "otherhost" in content
def test_bracketed_entry(self, temp_known_hosts):
"""Test removing bracketed entry like [hostname]:port."""
"""Test removing bracketed entry like [hostname]:port.
Note: ssh-keygen -R requires the exact hostname format, so bracketed
entries must be passed as "[hostname]:port", not just "hostname".
"""
existing = """[myhost]:2222 ssh-ed25519 AAAA...
otherhost ssh-rsa AAAA...
"""
Path(temp_known_hosts).write_text(existing)
result = remove_known_hosts_entry(temp_known_hosts, ["myhost"])
# Must use exact format for bracketed entries
result = remove_known_hosts_entry(temp_known_hosts, ["[myhost]:2222"])
assert result == 1
content = Path(temp_known_hosts).read_text()
@@ -853,13 +858,18 @@ otherhost ssh-rsa AAAA...
assert "otherhost" in content
def test_bracketed_ip_entry(self, temp_known_hosts):
"""Test removing bracketed IP entry like [192.168.1.1]:2222."""
"""Test removing bracketed IP entry like [192.168.1.1]:2222.
Note: ssh-keygen -R requires the exact hostname format, so bracketed
entries must be passed as "[ip]:port", not just "ip".
"""
existing = """[192.168.1.100]:2222 ssh-ed25519 AAAA...
otherhost ssh-rsa AAAA...
"""
Path(temp_known_hosts).write_text(existing)
result = remove_known_hosts_entry(temp_known_hosts, ["192.168.1.100"])
# Must use exact format for bracketed entries
result = remove_known_hosts_entry(temp_known_hosts, ["[192.168.1.100]:2222"])
assert result == 1
content = Path(temp_known_hosts).read_text()
@@ -867,20 +877,79 @@ otherhost ssh-rsa AAAA...
assert "otherhost" in content
def test_hashed_entries_preserved(self, temp_known_hosts):
"""Test that hashed entries (|1|...) are preserved."""
existing = """|1|abc123...= ssh-ed25519 AAAA...
myhost ssh-rsa AAAA...
|1|def456...= ssh-ed25519 AAAA...
"""Test that hashed entries (|1|...) are preserved when removing other hosts."""
import subprocess
# Create real hashed entries by hashing actual hostnames
existing = """other1.example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAITest1
myhost ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQTest
other2.example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAITest2
"""
Path(temp_known_hosts).write_text(existing)
# Hash the file to create real |1|... entries
subprocess.run(
["ssh-keygen", "-H", "-f", str(temp_known_hosts)],
capture_output=True,
check=True,
)
# Remove the .old backup file ssh-keygen creates
old_file = Path(temp_known_hosts + ".old")
if old_file.exists():
old_file.unlink()
# Verify we have hashed entries (all should be hashed now)
content = Path(temp_known_hosts).read_text()
lines = [line for line in content.strip().split("\n") if line]
assert len(lines) == 3
assert all(line.startswith("|1|") for line in lines)
# Now remove "myhost" - the other two hashed entries should remain
result = remove_known_hosts_entry(temp_known_hosts, ["myhost"])
assert result == 1
content = Path(temp_known_hosts).read_text()
assert "|1|abc123" in content
assert "|1|def456" in content
assert "myhost" not in content
lines = [line for line in content.strip().split("\n") if line]
# Should have exactly 2 entries remaining (other1 and other2)
assert len(lines) == 2
assert all(line.startswith("|1|") for line in lines)
def test_remove_hashed_entry(self, temp_known_hosts):
"""Test removing a hashed known_hosts entry by hostname.
This tests the actual bug: macOS/Linux often use HashKnownHosts yes,
which stores entries like |1|<salt>|<hash> instead of plaintext hostnames.
The function should still be able to remove these entries.
"""
import subprocess
hostname = "test.example.com"
# Create an unhashed entry first
unhashed = f"{hostname} ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAITest\n"
Path(temp_known_hosts).write_text(unhashed)
# Hash it using ssh-keygen -H
subprocess.run(
["ssh-keygen", "-H", "-f", str(temp_known_hosts)],
capture_output=True,
check=True,
)
# Remove the .old backup file ssh-keygen creates
old_file = Path(temp_known_hosts + ".old")
if old_file.exists():
old_file.unlink()
# Verify entry is now hashed
content = Path(temp_known_hosts).read_text()
assert content.startswith("|1|"), f"Entry should be hashed, got: {content}"
assert hostname not in content # Hostname should not appear in plaintext
# Now try to remove by hostname - this is the bug test
result = remove_known_hosts_entry(temp_known_hosts, [hostname])
assert result == 1, "Should have removed the hashed entry"
content = Path(temp_known_hosts).read_text()
assert content.strip() == "", f"File should be empty after removal, got: {content}"
def test_comments_preserved(self, temp_known_hosts):
"""Test that comments are preserved."""
@@ -942,17 +1011,14 @@ otherhost ssh-rsa AAAA...
remove_known_hosts_entry(temp_known_hosts, ["myhost"])
backup_path = Path(temp_known_hosts).parent / "known_hosts.bak"
# ssh-keygen -R creates .old backup (not .bak)
backup_path = Path(temp_known_hosts + ".old")
assert backup_path.exists()
backup_content = backup_path.read_text()
assert "myhost" in backup_content
assert "otherhost" in backup_content
# Check backup has same permissions
backup_mode = backup_path.stat().st_mode & 0o777
assert backup_mode == 0o600
def test_remove_all_entries(self, temp_known_hosts):
"""Test removing all entries from known_hosts."""
existing = """myhost ssh-ed25519 AAAA...