gib/Panama
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Show every answer the portal remembers, and give SSH keys their missing half

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ms2FbjQy31TVf3CEvQhGM8
This commit is contained in:
Gabriel Brown
2026-08-24 19:26:56 -04:00
parent 4ec8bd94d9
commit 6f0ce639d9
25 changed files with 3622 additions and 408 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/dot/quickshell/scripts/panama-permissions
+205 -63
View File
@@ -1,21 +1,40 @@
#!/usr/bin/env python3
"""Which applications may use the camera and microphone.
"""Which applications the desktop portal has recorded an answer for.
Read from xdg-desktop-portal's permission store, which is where an application
that asks through the portal has its answer recorded. That is the whole of what
this can control, and the limit is worth stating plainly rather than implying a
protection that does not exist: a native binary opens /dev/video0 directly and
no desktop setting stands in its way. What this covers is Flatpaks and anything
else that goes through the portal -- which on this machine is most of what would
ever ask.
that asks through the portal has its answer written down. That is the whole of
what this can control, and the limit is worth stating plainly rather than
implying a protection that does not exist: a native binary opens /dev/video0
directly and no desktop setting stands in its way. What this covers is Flatpaks
and anything else that goes through the portal -- which on this machine is most
of what would ever ask.
Devices with no recorded application are reported as empty rather than omitted,
Six subjects, mapped onto the store's own layout rather than onto a tidier one:
camera, microphone entries of the store's `devices` table, keyed by the
device name. Plain yes/no, so they can be toggled.
background the `background` table's single `background` entry, also
plain yes/no. This is "may run when you closed it".
screencast its own table, and remote-desktop likewise -- but their
remote-desktop ids are opaque restore tokens, one per remembered
session, and the value beside each grant is a structured
GVariant describing which monitor or which input devices
were shared. Nothing here can rebuild one of those, so
these report which application holds grants and offer
only to drop them. There is deliberately no code path
that Sets them: a toggle that cannot be honoured is worse
than no toggle.
location listed only. geoclue is absent on this machine, so the
table is normally empty and the page omits the section
rather than showing an empty one.
A table nothing has asked for is reported as an empty list rather than omitted,
so the page can say "nothing has asked" instead of showing nothing at all.
panama-permissions snapshot
panama-permissions set DEVICE APP_ID allow|deny
panama-permissions forget DEVICE APP_ID
panama-permissions set TABLE APP_ID true|false
panama-permissions forget TABLE APP_ID
"""
from __future__ import annotations
@@ -24,24 +43,58 @@ import json
import re
import subprocess
import sys
TABLE = "devices"
# The devices the portal arbitrates. Listed rather than discovered so a device
# nothing has asked for still appears, which is the difference between "no
# application uses your microphone" and a page that silently omits it.
DEVICES = (
("camera", "Camera"),
("microphone", "Microphone"),
("speakers", "Speakers"),
)
from dataclasses import dataclass
APP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
DEVICE_ID = re.compile(r"^[a-z]+$")
TABLE_ID = re.compile(r"^[a-z][a-z-]{0,31}$")
# The store's own ids for a remembered session are opaque tokens, so they are
# never accepted from a caller -- only discovered by listing the table.
ENTRY_ID = re.compile(r"^[A-Za-z0-9._-]{1,128}$")
ALLOWED = "yes"
DENIED = "no"
SIMPLE = "simple"
STRUCTURED = "structured"
READONLY = "readonly"
@dataclass(frozen=True)
class Table:
"""One subject on the page, and where the store keeps it.
`portal` is the store's table name, which is not always the subject's name:
camera and microphone both live in `devices`. `entries` names the store ids
to look under, or None to discover them by listing the table -- which is the
only option for screencast and remote-desktop, whose ids are restore tokens.
`writable` is the id a Set writes to, and is None for everything this
refuses to write.
"""
name: str
portal: str
entries: tuple[str, ...] | None
writable: str | None
kind: str
TABLES: tuple[Table, ...] = (
Table("camera", "devices", ("camera",), "camera", SIMPLE),
Table("microphone", "devices", ("microphone",), "microphone", SIMPLE),
Table("screencast", "screencast", None, None, STRUCTURED),
Table("remote-desktop", "remote-desktop", None, None, STRUCTURED),
Table("background", "background", None, "background", SIMPLE),
Table("location", "location", None, None, READONLY),
)
BY_NAME = {table.name: table for table in TABLES}
# Named once, here, so the refusal below reads as a list membership rather than
# as a chain of conditions someone could later add an exception to.
SIMPLE_TABLES = tuple(table.name for table in TABLES if table.kind == SIMPLE)
REVOKE_ONLY_TABLES = tuple(table.name for table in TABLES if table.kind == STRUCTURED)
class BoundaryError(RuntimeError):
"""A user-visible validation or permission-store failure."""
@@ -71,7 +124,7 @@ def portal_call(method: str, signature: str, *arguments: str) -> dict | None:
])
if result.returncode != 0:
detail = (result.stderr or "").strip()
# A device nothing has ever asked for has no row at all, and the store
# An entry nothing has ever asked for has no row at all, and the store
# says so as "No entry for camera". Matched on the store's actual words
# rather than a guess at them.
lowered = detail.lower()
@@ -89,44 +142,91 @@ def available() -> bool:
return "org.freedesktop.impl.portal.PermissionStore" in (result.stdout or "")
def entries_for(device: str) -> list[dict]:
payload = portal_call("Lookup", "ss", TABLE, device)
def entry_ids(table: Table) -> list[str]:
"""The store ids to look under for one subject.
Fixed for the device entries, discovered for everything else. A discovered
id that does not look like an id is dropped rather than passed back into the
store.
"""
if table.entries is not None:
return list(table.entries)
payload = portal_call("List", "s", table.portal)
if not payload:
return []
data = payload.get("data") or []
if not data or not isinstance(data[0], dict):
return []
listed = data[0] if data and isinstance(data[0], list) else []
return [str(value) for value in listed if ENTRY_ID.match(str(value))]
found = []
for app_id, permissions in data[0].items():
values = [str(value) for value in (permissions or [])]
found.append({
"app": app_id,
# Anything that is not an explicit "yes" is treated as withheld:
# guessing generously about a camera is the wrong way to be wrong.
"allowed": ALLOWED in values,
"raw": ",".join(values),
})
found.sort(key=lambda entry: entry["app"].casefold())
return found
def rows_for(table: Table) -> list[dict]:
"""One row per application, folded across every entry in the table.
screencast keeps a separate entry per remembered session, so an application
that has shared its screen four times appears four times in the store. The
page is answering "may this application share your screen", which is one
question, so the rows are folded by application and `grants` says how many
stored sessions are behind the row.
"""
folded: dict[str, dict] = {}
for entry in entry_ids(table):
payload = portal_call("Lookup", "ss", table.portal, entry)
if not payload:
continue
data = payload.get("data") or []
recorded = data[0] if data and isinstance(data[0], dict) else {}
if not isinstance(recorded, dict):
continue
for app_id, permissions in recorded.items():
values = [str(value) for value in (permissions or [])]
row = folded.setdefault(str(app_id), {
"app": str(app_id),
# Anything that is not an explicit "yes" is treated as withheld:
# guessing generously about a camera is the wrong way to be wrong.
"allowed": False,
"grants": 0,
"raw": "",
})
row["grants"] += 1
if ALLOWED in values:
row["allowed"] = True
if row["raw"] == "":
row["raw"] = ",".join(values)
rows = list(folded.values())
rows.sort(key=lambda row: row["app"].casefold())
return rows
def snapshot() -> dict:
if not available():
return {
"available": False,
"devices": [],
"tables": {table.name: [] for table in TABLES},
"simpleTables": list(SIMPLE_TABLES),
"revokeOnlyTables": list(REVOKE_ONLY_TABLES),
"error": "The desktop portal's permission store is not running.",
}
devices = []
for device_id, label in DEVICES:
devices.append({
"id": device_id,
"label": label,
"applications": entries_for(device_id),
})
return {"available": True, "devices": devices, "error": ""}
tables: dict[str, list[dict]] = {}
problems: list[str] = []
for table in TABLES:
# One unreadable table must not take the other five down with it: a page
# that shows nothing because location is broken is worse than a page
# that shows five subjects and says location could not be read.
try:
tables[table.name] = rows_for(table)
except BoundaryError as error:
tables[table.name] = []
problems.append(f"{table.name}: {error}")
return {
"available": True,
"tables": tables,
"simpleTables": list(SIMPLE_TABLES),
"revokeOnlyTables": list(REVOKE_ONLY_TABLES),
"error": "; ".join(problems),
}
def require(pattern: re.Pattern[str], value: str, message: str) -> str:
@@ -135,22 +235,59 @@ def require(pattern: re.Pattern[str], value: str, message: str) -> str:
return value
def set_permission(device: str, app: str, allowed: bool) -> None:
require(DEVICE_ID, device, "That is not a device.")
def named(name: str) -> Table:
require(TABLE_ID, name, "That is not a permission table.")
table = BY_NAME.get(name)
if table is None:
raise BoundaryError("That is not a permission table this manages.")
return table
def set_permission(name: str, app: str, allowed: bool) -> None:
"""Record an answer. Only for the subjects whose value is a plain yes or no.
screencast and remote-desktop are refused here and have no other route to a
Set anywhere in this file. Their stored value is a structured description of
a session -- which monitor, which input devices -- and writing a bare "yes"
over it would leave the store holding a grant the portal cannot restore.
Dropping the grant is the honest operation, so that is the only one offered.
"""
table = named(name)
require(APP_ID, app, "That is not an application.")
if not any(device == known for known, _ in DEVICES):
raise BoundaryError("That is not a device this manages.")
if table.name not in SIMPLE_TABLES or table.writable is None:
if table.kind == STRUCTURED:
raise BoundaryError(
f"A {table.name} grant describes a whole session, so it can only "
"be revoked, not switched on and off.")
raise BoundaryError(f"{table.name} permissions are shown, not changed, here.")
# Permissions are an array of strings, so busctl needs the element count
# before the element -- "1 yes", not "yes".
portal_call("SetPermission", "sbssas", TABLE, "true", device, app,
portal_call("SetPermission", "sbssas", table.portal, "true", table.writable, app,
"1", ALLOWED if allowed else DENIED)
def forget(device: str, app: str) -> None:
"""Drop the recorded answer, so the application is asked again next time."""
require(DEVICE_ID, device, "That is not a device.")
def forget(name: str, app: str) -> None:
"""Drop the recorded answer, so the application is asked again next time.
Every entry the application appears under, because one application can hold
several remembered screencast sessions and dropping one of them would leave
the row on the page looking unchanged.
"""
table = named(name)
require(APP_ID, app, "That is not an application.")
portal_call("DeletePermission", "sss", TABLE, device, app)
dropped = 0
for entry in entry_ids(table):
payload = portal_call("Lookup", "ss", table.portal, entry)
if not payload:
continue
data = payload.get("data") or []
recorded = data[0] if data and isinstance(data[0], dict) else {}
if not isinstance(recorded, dict) or app not in recorded:
continue
portal_call("DeletePermission", "sss", table.portal, entry, app)
dropped += 1
if dropped == 0:
raise BoundaryError("There was no recorded answer to forget.")
def main(arguments: list[str]) -> int:
@@ -160,20 +297,25 @@ def main(arguments: list[str]) -> int:
return 0
if len(arguments) == 4 and arguments[0] == "set":
if arguments[3] not in ("allow", "deny"):
raise BoundaryError("That is not allow or deny.")
set_permission(arguments[1], arguments[2], arguments[3] == "allow")
if arguments[3] not in ("true", "false"):
raise BoundaryError("That is not true or false.")
set_permission(arguments[1], arguments[2], arguments[3] == "true")
elif len(arguments) == 3 and arguments[0] == "forget":
forget(arguments[1], arguments[2])
else:
raise BoundaryError(
"Usage: panama-permissions snapshot | set DEVICE APP allow|deny | "
"forget DEVICE APP")
"Usage: panama-permissions snapshot | set TABLE APP true|false | "
"forget TABLE APP")
except BoundaryError as error:
try:
state = snapshot()
except BoundaryError:
state = {"available": False, "devices": []}
state = {
"available": False,
"tables": {table.name: [] for table in TABLES},
"simpleTables": list(SIMPLE_TABLES),
"revokeOnlyTables": list(REVOKE_ONLY_TABLES),
}
state["error"] = str(error)
print(json.dumps(state, separators=(",", ":")))
return 0
config/dot/quickshell/scripts/panama-privacy
Executable
+269
View File
@@ -0,0 +1,269 @@
#!/usr/bin/env python3
"""What this desktop remembers about what you opened, and how to forget it.
Two traces, both of them a normal and useful part of a desktop rather than a
problem to be alarmed about. This tool measures them and clears them on request.
It never clears anything on its own, never runs on a schedule, and reports sizes
without ever suggesting that a number is too big -- the software that tells you
your machine is dirty is trying to sell you something.
traces what each trace currently costs, measured now.
clear-recents the recent-files list, replaced with an empty but valid
document. Not deleted: GTK recreates the file the moment
something opens a file anyway, and an empty valid file takes
effect in every running application immediately, where a
missing one is only noticed on the next write.
clear-thumbnails the contents of the thumbnail cache. The folder stays; only
what is inside it goes, and nothing is followed out of it.
Trash is deliberately not here. It is measured and emptied by panama-disks, and
one trash implementation is the right number to have.
Seams, for tests that must not touch a real home directory:
PANAMA_PRIVACY_RECENTS the recent-files document
PANAMA_PRIVACY_THUMBNAILS the thumbnail cache folder
Both still have to resolve to somewhere inside HOME, so a test points HOME at a
scratch directory rather than pointing these at one.
panama-privacy traces
panama-privacy clear-recents
panama-privacy clear-thumbnails
"""
from __future__ import annotations
import json
import os
import re
import shutil
import subprocess
import sys
import time
from pathlib import Path
# Measuring a thumbnail cache means walking it, and a machine that has browsed a
# large picture library has a lot of it. Bounded so the page cannot hang; a walk
# that runs out of time reports what it had and says it is a floor.
MEASURE_TIMEOUT_SECONDS = 20.0
# The document GTK keeps the recent-files list in, and the cache every file
# manager and image viewer on this desktop shares.
RECENTS_RELATIVE = "recently-used.xbel"
THUMBNAILS_RELATIVE = "thumbnails"
# What an emptied recent-files list looks like. Byte for byte the header GTK
# writes itself, so the file this leaves behind is one GTK would have written.
EMPTY_XBEL = (
'<?xml version="1.0" encoding="UTF-8"?>\n'
'<xbel version="1.0"\n'
' xmlns:bookmark="http://www.freedesktop.org/standards/desktop-bookmarks"\n'
' xmlns:mime="http://www.freedesktop.org/standards/shared-mime-info"\n'
'>\n'
'</xbel>\n'
)
BOOKMARK = re.compile(rb"<bookmark\b")
class BoundaryError(RuntimeError):
"""A user-visible validation or filesystem failure."""
def home() -> Path:
return Path(os.path.expanduser("~")).resolve(strict=False)
def data_home() -> Path:
configured = os.environ.get("XDG_DATA_HOME") or ""
if configured:
return Path(configured)
return Path(os.path.expanduser("~")) / ".local" / "share"
def cache_home() -> Path:
configured = os.environ.get("XDG_CACHE_HOME") or ""
if configured:
return Path(configured)
return Path(os.path.expanduser("~")) / ".cache"
def recents_path() -> Path:
override = os.environ.get("PANAMA_PRIVACY_RECENTS") or ""
return Path(override) if override else data_home() / RECENTS_RELATIVE
def thumbnails_path() -> Path:
override = os.environ.get("PANAMA_PRIVACY_THUMBNAILS") or ""
return Path(override) if override else cache_home() / THUMBNAILS_RELATIVE
def confined(target: Path, description: str) -> Path:
"""A path this is allowed to write to, or a refusal.
The same guard panama-disks uses on the cache folder, and for the same
reason: this function is the whole reason the buttons on the page are safe
to press. A symlink is refused outright rather than followed, and anything
that resolves to the home directory itself or to somewhere outside it is
refused -- so an XDG variable pointing somewhere alarming, or a cache folder
someone linked to /, cannot turn one click into a deleted system.
"""
if target.is_symlink():
raise BoundaryError(f"{description} is a link, so it will not be touched.")
resolved = target.resolve(strict=False)
root = home()
if resolved == root or root not in resolved.parents:
raise BoundaryError(f"{description} is not inside your home folder.")
return resolved
def measure(path: Path, budget: float) -> tuple[int, float, bool]:
"""Bytes used, what is left of the budget, and whether the walk finished."""
if not path.exists():
return 0, budget, True
if budget <= 1.0:
return 0, budget, False
started = time.monotonic()
try:
completed = subprocess.run(["du", "-sxb", str(path)], check=False,
capture_output=True, text=True, timeout=budget)
except (OSError, subprocess.TimeoutExpired):
# A walk that ran out of time has consumed the whole budget by
# definition; the caller stops rather than starting another.
return 0, 0.0, False
left = max(budget - (time.monotonic() - started), 0.0)
if completed.returncode != 0:
return 0, left, False
match = re.match(r"^(\d+)", completed.stdout)
return (int(match.group(1)) if match else 0), left, match is not None
def recent_entries(path: Path) -> int:
"""How many files the recent list remembers.
Counted by scanning for the opening tag rather than parsing the document:
the answer wanted is a count, and an XML parser here would build a tree of
every path the user has opened in order to throw it away again.
"""
try:
with path.open("rb") as document:
return sum(len(BOOKMARK.findall(chunk))
for chunk in iter(lambda: document.read(65536), b""))
except OSError:
return 0
def traces() -> dict:
budget = MEASURE_TIMEOUT_SECONDS
recents = recents_path()
recents_present = recents.is_file()
recents_bytes = recents.stat().st_size if recents_present else 0
thumbnails = thumbnails_path()
thumbnails_present = thumbnails.is_dir()
thumbnails_bytes, budget, complete = measure(thumbnails, budget)
return {
"recents": {
"bytes": int(recents_bytes),
"entries": recent_entries(recents) if recents_present else 0,
"path": str(recents),
"present": recents_present,
},
"thumbnails": {
"bytes": int(thumbnails_bytes),
"path": str(thumbnails),
"present": thumbnails_present,
# False when the walk ran out of time, which makes the byte count a
# floor rather than an answer. The page says so instead of quoting a
# number it cannot stand behind.
"measured": complete,
},
"error": "",
}
def clear_recents() -> None:
"""Replace the recent-files list with an empty one.
Never unlinked. GTK holds the path open and recreates the document on its
next write, so deleting it buys nothing an empty document does not, and an
empty document is understood by everything reading the list right now.
"""
target = recents_path()
if not target.exists():
# Nothing remembered is the state this was asked to produce.
return
if not target.is_file():
raise BoundaryError("The recent-files list is not a file.")
resolved = confined(target, "The recent-files list")
try:
resolved.write_text(EMPTY_XBEL, encoding="utf-8")
except OSError as error:
raise BoundaryError("The recent-files list could not be emptied.") from error
def clear_thumbnails() -> None:
"""Delete what is inside the thumbnail cache, never following a link out.
A thumbnail an application still has open cannot always be removed, and that
is the normal case rather than a failure -- so a partial pass succeeds, and
the freshly measured size the caller gets back says how much is left. Only a
pass that removed nothing at all is reported as a failure.
"""
target = thumbnails_path()
if not target.exists():
return
if not target.is_dir():
raise BoundaryError("The thumbnail cache is not a folder.")
directory = confined(target, "The thumbnail cache")
removed = 0
failures = 0
with os.scandir(directory) as entries:
for entry in entries:
try:
# is_symlink first: a symlinked directory must be unlinked, not
# walked, or this deletes whatever it points at.
if entry.is_symlink() or not entry.is_dir(follow_symlinks=False):
os.unlink(entry.path)
else:
# rmtree lstats as it goes and refuses to descend a symlink.
shutil.rmtree(entry.path, ignore_errors=False)
removed += 1
except OSError:
failures += 1
if failures and not removed:
raise BoundaryError("The thumbnail cache is in use and nothing could be removed.")
def main(arguments: list[str]) -> int:
try:
if arguments == ["traces"]:
pass
elif arguments == ["clear-recents"]:
clear_recents()
elif arguments == ["clear-thumbnails"]:
clear_thumbnails()
else:
raise BoundaryError(
"Usage: panama-privacy traces | clear-recents | clear-thumbnails")
except BoundaryError as error:
try:
state = traces()
except OSError:
state = {"recents": {"bytes": 0, "entries": 0, "path": "", "present": False},
"thumbnails": {"bytes": 0, "path": "", "present": False, "measured": False}}
state["error"] = str(error)
print(json.dumps(state, separators=(",", ":")))
return 0
print(json.dumps(traces(), separators=(",", ":")))
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))
config/dot/quickshell/scripts/panama-ssh-keys
+278 -7
View File
@@ -8,13 +8,21 @@ ssh-keygen to derive the PUBLIC key with an empty passphrase: it succeeds for an
unencrypted key and fails for an encrypted one, and either way the only thing it
can print is public material.
No passphrase passes through this tool at all. Adding an encrypted key to the
agent lets ssh-add prompt through the system's own askpass, which is where that
belongs -- a settings page collecting a passphrase and handing it on would be a
worse place for it to live, and putting one in argv would publish it to every
process on the machine.
A passphrase enters this tool in exactly one place -- creating a key -- and it
enters on standard input, which no other process can read. From there it is
typed at ssh-keygen over a pseudo-terminal, the same way a person would type it,
because the two obvious alternatives are both worse: a passphrase in argv is
published to every process on the machine, and a passphrase in a temporary file
is written to disk. It is never logged, never echoed back, and never included in
an error message.
Adding an existing encrypted key to the agent is different: no passphrase is
collected for that at all, because ssh-add prompts through the system's own
askpass, which is where that belongs.
panama-ssh-keys snapshot
panama-ssh-keys generate NAME COMMENT (passphrase on stdin)
panama-ssh-keys fix-permissions NAME
panama-ssh-keys agent-add PATH | agent-remove PATH
panama-ssh-keys forget-host HOST
"""
@@ -23,9 +31,14 @@ from __future__ import annotations
import json
import os
import pty
import re
import select
import shutil
import signal
import subprocess
import sys
import time
from pathlib import Path
SSH_DIR = Path.home() / ".ssh"
@@ -34,6 +47,23 @@ KNOWN_HOSTS = SSH_DIR / "known_hosts"
# A host as it may appear in known_hosts, including [host]:port forms.
HOST = re.compile(r"^[A-Za-z0-9._:\[\]-]{1,253}$")
# A key file name, and nothing that could be a path. No slash is in the class,
# so a name cannot describe another directory at all -- the resolve-and-compare
# below is the second lock on the same door rather than the only one.
KEY_NAME = re.compile(r"^[A-Za-z0-9_.-]{1,64}$")
# A key comment. Free text, but nothing that could break out of a terminal line
# or be mistaken for one of ssh-keygen's own prompts.
KEY_COMMENT = re.compile(r"^[^\x00-\x1f\x7f]{0,128}$")
# ssh-keygen's own floor. Checked here so the refusal arrives before a terminal
# is opened, rather than as a re-prompt nobody is there to answer.
MINIMUM_PASSPHRASE = 5
# Generating an ed25519 key takes milliseconds. The budget is this large only so
# that a machine starved of entropy fails with a message rather than a hang.
KEYGEN_TIMEOUT_SECONDS = 120.0
# gnome-keyring's agent, which is what runs on this desktop. Only used when the
# environment has not already named one, so an ssh-agent started by hand wins.
KEYRING_SOCKET = Path(os.environ.get("XDG_RUNTIME_DIR", "/run/user/1000")) / "keyring" / "ssh"
@@ -253,6 +283,217 @@ def agent_remove(path: str) -> None:
raise BoundaryError(detail[-1] if detail else "That key could not be removed.")
def resolve_new_key(name: str) -> Path:
"""Where a key by this name would go, or a refusal.
Refuses anything that already exists -- both halves, because a stray .pub
beside no private key still means ssh-keygen would be asked to overwrite,
and this tool does not overwrite keys. Losing a private key is not
recoverable and a settings page is the wrong place to learn that.
"""
if not KEY_NAME.match(name or "") or name in (".", ".."):
raise BoundaryError(
"A key name can use letters, numbers, dots, dashes and underscores.")
if name.endswith(".pub"):
raise BoundaryError("Name the key itself, not its public half.")
try:
SSH_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
directory = SSH_DIR.resolve(strict=True)
except OSError as error:
raise BoundaryError("The SSH directory could not be opened.") from error
if not directory.is_dir():
raise BoundaryError("The SSH directory is not a directory.")
target = directory / name
if target.parent.resolve(strict=True) != directory:
raise BoundaryError("That key is not in the SSH directory.")
public = Path(str(target) + ".pub")
for candidate in (target, public):
if candidate.exists() or candidate.is_symlink():
raise BoundaryError(f"{candidate.name} already exists, so nothing was written.")
return target
def terminal_environment() -> dict:
"""The environment ssh-keygen must run in to ask its question at the terminal.
This desktop sets SSH_ASKPASS_REQUIRE=prefer, which makes ssh-keygen open a
graphical passphrase dialog even when it has a perfectly good terminal in
front of it -- so the first version of this hung, waiting for a prompt that
had been drawn on somebody's screen instead. The terminal is supplied
deliberately here, so the askpass route is switched off just as deliberately.
LC_ALL is pinned so the prompts read below are the ones OpenSSH ships.
"""
environment = dict(os.environ)
environment["SSH_ASKPASS_REQUIRE"] = "never"
environment["LC_ALL"] = "C"
for name in ("SSH_ASKPASS", "DISPLAY", "WAYLAND_DISPLAY"):
environment.pop(name, None)
return environment
def type_at_keygen(command: list[str], passphrase: str) -> None:
"""Run ssh-keygen on a pseudo-terminal and answer its prompts.
ssh-keygen reads a passphrase through readpassphrase(), which opens
/dev/tty: a pipe on standard input is not read at all, which is why this
needs a terminal rather than a simpler subprocess call. The passphrase is
written to the terminal's master side, exactly as typing it would, and
ssh-keygen asks twice, so it is typed twice.
Nothing about the passphrase is kept. It is not written to disk, does not
appear in the command, and is scrubbed out of anything reported back in case
a future ssh-keygen ever echoes it.
"""
try:
pid, master = pty.fork()
except OSError as error:
raise BoundaryError("A terminal could not be opened for ssh-keygen.") from error
if pid == 0:
# The child. Nothing may return from here into the parent's code holding
# the parent's file descriptors, so a failed exec exits outright.
try:
os.execvpe(command[0], command, terminal_environment())
except OSError:
pass
os._exit(127)
typed = 0
pending = ""
transcript = ""
problem = ""
deadline = time.monotonic() + KEYGEN_TIMEOUT_SECONDS
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
problem = "ssh-keygen did not finish."
break
try:
ready, _, _ = select.select([master], [], [], min(remaining, 1.0))
except OSError:
break
if not ready:
continue
try:
chunk = os.read(master, 4096)
except OSError:
# EIO on Linux: the child closed the terminal, which is how a pty
# reports end of output.
break
if not chunk:
break
text = chunk.decode("utf-8", errors="replace")
pending += text
transcript += text
lowered = pending.lower()
# Matched on ssh-keygen's whole prompt rather than one word of it: the
# passphrase prompt quotes the key's path back, and "overwrite" is a
# perfectly legal key name.
if "overwrite (y/n)" in lowered:
# Unreachable in practice -- an existing key is refused before this
# runs -- but answering anything other than "no" here would destroy
# a key, so it answers no and stops.
os.write(master, b"n\n")
problem = "That key already exists, so nothing was written."
break
if "passphrase is too short" in lowered:
problem = (f"ssh-keygen wants a passphrase of at least "
f"{MINIMUM_PASSPHRASE} characters.")
break
if "passphrases do not match" in lowered:
problem = "Those passphrases did not match."
break
if typed < 2 and "passphrase" in lowered and pending.rstrip().endswith(":"):
os.write(master, passphrase.encode("utf-8") + b"\n")
typed += 1
pending = ""
if problem:
try:
os.kill(pid, signal.SIGKILL)
except OSError:
pass
try:
os.close(master)
except OSError:
pass
try:
_, status = os.waitpid(pid, 0)
except OSError:
status = 0
if problem:
raise BoundaryError(problem)
if not (os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0):
detail = scrubbed(transcript, passphrase)
last = [line.strip() for line in detail.splitlines() if line.strip()]
raise BoundaryError(last[-1] if last else "ssh-keygen could not create that key.")
def scrubbed(text: str, secret: str) -> str:
return text.replace(secret, "********") if secret else text
def generate(name: str, comment: str, passphrase: str, allow_empty: bool) -> None:
"""Create an ed25519 key.
ed25519 and nothing else: it is the key everything current accepts, the
choice between it and RSA is not one a settings page should make someone
make, and offering a size field for a curve that has one size would be
theatre.
"""
if not KEY_COMMENT.match(comment or ""):
raise BoundaryError("A comment cannot contain control characters.")
if "\n" in passphrase or "\r" in passphrase:
raise BoundaryError("A passphrase cannot contain a line break.")
if passphrase == "":
if not allow_empty:
raise BoundaryError(
"A passphrase is required. A key with none is usable by anyone "
"who reads the file.")
elif len(passphrase) < MINIMUM_PASSPHRASE:
raise BoundaryError(
f"ssh-keygen wants a passphrase of at least {MINIMUM_PASSPHRASE} characters.")
if shutil.which("ssh-keygen") is None:
raise BoundaryError("ssh-keygen is not installed.")
target = resolve_new_key(name)
command = ["ssh-keygen", "-t", "ed25519", "-f", str(target)]
if comment:
command += ["-C", comment]
type_at_keygen(command, passphrase)
if not target.is_file() or not Path(str(target) + ".pub").is_file():
raise BoundaryError("ssh-keygen finished but the key is not there.")
def fix_permissions(name: str) -> None:
"""Make a private key readable only by its owner.
ssh refuses to use a key other people can read, and says so in a message
most people meet for the first time at the worst moment. The path is
resolved and compared against the SSH directory first, so a name that is a
link to something elsewhere is refused rather than followed -- this changes
a file's mode, and that is not a thing to do to a file you have not checked.
"""
if not KEY_NAME.match(name or "") or name in (".", ".."):
raise BoundaryError("That is not a key name.")
key = resolve_key(str(SSH_DIR / name))
try:
os.chmod(key, 0o600)
except OSError as error:
raise BoundaryError("That key's permissions could not be changed.") from error
def forget_host(host: str) -> None:
"""Drop a host's keys from known_hosts.
@@ -271,13 +512,42 @@ def forget_host(host: str) -> None:
raise BoundaryError(detail[-1] if detail else "That host could not be removed.")
def read_passphrase() -> str:
"""The passphrase, from standard input, and only from there.
One trailing newline is dropped because the caller writes one to end the
line; anything else is taken literally, including spaces, because a
passphrase is allowed to end in one.
"""
try:
raw = sys.stdin.buffer.read().decode("utf-8")
except (OSError, UnicodeDecodeError) as error:
raise BoundaryError("The passphrase could not be read.") from error
if raw.endswith("\n"):
raw = raw[:-1]
if raw.endswith("\r"):
raw = raw[:-1]
return raw
def main(arguments: list[str]) -> int:
try:
if arguments == ["snapshot"]:
print(json.dumps(snapshot(), separators=(",", ":")))
return 0
if len(arguments) == 2 and arguments[0] == "agent-add":
if arguments and arguments[0] == "generate":
# The flag exists so a machine-shaped caller can ask for a key with
# no passphrase deliberately. Panama's own page never passes it: it
# requires a passphrase and validates that both fields match.
allow_empty = "--no-passphrase" in arguments[1:]
rest = [value for value in arguments[1:] if value != "--no-passphrase"]
if len(rest) != 2:
raise BoundaryError("Usage: panama-ssh-keys generate NAME COMMENT")
generate(rest[0], rest[1], read_passphrase(), allow_empty)
elif len(arguments) == 2 and arguments[0] == "fix-permissions":
fix_permissions(arguments[1])
elif len(arguments) == 2 and arguments[0] == "agent-add":
agent_add(arguments[1])
elif len(arguments) == 2 and arguments[0] == "agent-remove":
agent_remove(arguments[1])
@@ -285,7 +555,8 @@ def main(arguments: list[str]) -> int:
forget_host(arguments[1])
else:
raise BoundaryError(
"Usage: panama-ssh-keys snapshot | agent-add PATH | agent-remove PATH | "
"Usage: panama-ssh-keys snapshot | generate NAME COMMENT | "
"fix-permissions NAME | agent-add PATH | agent-remove PATH | "
"forget-host HOST")
except BoundaryError as error:
try: