50 lines
2.1 KiB
Python
Executable File
50 lines
2.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Install a verified Ananke peer public key for atlas, restricted to Titan-24.
|
|
|
|
Read {"hostname":"titan-12", "public_key":"ssh-ed25519 ..."} from stdin.
|
|
Obtain the key from the authenticated peer, never from an unverified key scan.
|
|
"""
|
|
import base64
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import pwd
|
|
import socket
|
|
import sys
|
|
|
|
|
|
def install(payload):
|
|
"""Append one public key without replacing existing administrator keys."""
|
|
if os.geteuid() != 0 or payload["hostname"] != socket.gethostname().split(".")[0]:
|
|
raise ValueError("root_and_matching_hostname_required")
|
|
parts = payload["public_key"].split()
|
|
if len(parts) < 2 or parts[0] != "ssh-ed25519":
|
|
raise ValueError("ed25519_public_key_required")
|
|
blob = base64.b64decode(parts[1], validate=True)
|
|
if len(blob) != 51 or not blob.startswith(b"\x00\x00\x00\x0bssh-ed25519\x00\x00\x00\x20"):
|
|
raise ValueError("invalid_public_key")
|
|
account = pwd.getpwnam("atlas")
|
|
directory = Path(account.pw_dir) / ".ssh"
|
|
target = directory / "authorized_keys"
|
|
if directory.is_symlink() or target.is_symlink():
|
|
raise ValueError("symlink_requires_review")
|
|
directory.mkdir(exist_ok=True, mode=0o700)
|
|
directory.chmod(0o700)
|
|
os.chown(directory, account.pw_uid, account.pw_gid)
|
|
old = target.read_text() if target.exists() else ""
|
|
key = " ".join(parts[:2])
|
|
if any(key in line for line in old.splitlines() if not line.lstrip().startswith("#")):
|
|
return {"hostname": payload["hostname"], "key_already_present": True}
|
|
prefix = "\n" if old and not old.endswith("\n") else ""
|
|
entry = prefix + 'from="192.168.22.26",restrict ' + key + " ananke-tethys-recovery\n"
|
|
fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_APPEND | os.O_NOFOLLOW, 0o600)
|
|
with os.fdopen(fd, "w") as stream:
|
|
os.fchmod(stream.fileno(), 0o600)
|
|
os.fchown(stream.fileno(), account.pw_uid, account.pw_gid)
|
|
stream.write(entry)
|
|
return {"hostname": payload["hostname"], "key_added": True, "source_ip": "192.168.22.26"}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(json.dumps(install(json.load(sys.stdin))))
|