mirror of
https://github.com/Such-Software/wownerogue.git
synced 2026-10-22 08:38:33 +02:00
1647 lines
54 KiB
Python
Executable File
1647 lines
54 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Fixed, root-only encrypted-wallet keyfile/password recovery proof for ``deb``.
|
|
|
|
The two accepted profile names select compile-time paths. No caller-controlled path, port, unit,
|
|
binary, user, or RPC method is accepted. The proof reads no mnemonic and has no mnemonic RPC.
|
|
Its only output is ``result=passed`` or ``result=refused``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ctypes
|
|
import grp
|
|
import hashlib
|
|
import hmac
|
|
import http.client
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import pwd
|
|
import resource
|
|
import signal
|
|
import socket
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
from typing import Callable
|
|
|
|
|
|
PR_SET_DUMPABLE = 4
|
|
PR_SET_KEEPCAPS = 8
|
|
PR_SET_NO_NEW_PRIVS = 38
|
|
MADV_DONTDUMP = 16
|
|
O_DIRECTORY = getattr(os, "O_DIRECTORY", 0)
|
|
O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
|
|
O_CLOEXEC = getattr(os, "O_CLOEXEC", 0)
|
|
MAX_CMDLINE = 64 * 1024
|
|
MAX_RPC_RESPONSE = 1024 * 1024
|
|
RPC_PATH = "/json_rpc"
|
|
RPC_USER = "recovery-proof"
|
|
ZERO_BLOCK = b"\0" * (1024 * 1024)
|
|
LIBC = ctypes.CDLL(None, use_errno=True)
|
|
BASE58_ALPHABET = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
|
|
|
|
ADDRESS_REQUEST = json.dumps(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": "keyfile-recovery-proof",
|
|
"method": "get_address",
|
|
"params": {"account_index": 0},
|
|
},
|
|
separators=(",", ":"),
|
|
).encode("ascii")
|
|
SIGN_MESSAGE = "wowngeon-wallet-keyfile-recovery-proof-v1"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WalletProfile:
|
|
name: str
|
|
unit: str
|
|
user: str
|
|
group: str
|
|
binary: pathlib.Path
|
|
live_wallet: pathlib.Path
|
|
live_rpc_port: int
|
|
live_bind_address: str
|
|
snapshot_wallet: pathlib.Path
|
|
snapshot_keys: pathlib.Path
|
|
proof_port: int
|
|
stagenet: bool
|
|
address_kind: str
|
|
proof_directory: pathlib.Path
|
|
|
|
@property
|
|
def candidate_directory(self) -> pathlib.Path:
|
|
return self.proof_directory / "candidate"
|
|
|
|
|
|
SNAPSHOT_DIRECTORY = pathlib.Path("/var/backups/wowngeon/wallets")
|
|
PROFILES = {
|
|
"monero-stagenet": WalletProfile(
|
|
name="monero-stagenet",
|
|
unit="monero-wallet-rpc.service",
|
|
user="jw",
|
|
group="jw",
|
|
binary=pathlib.Path("/usr/bin/monero-wallet-rpc"),
|
|
live_wallet=pathlib.Path(
|
|
"/home/jw/Programs/monero-x86_64-linux-gnu-v0.18.4.4/test"
|
|
),
|
|
live_rpc_port=38083,
|
|
live_bind_address="127.0.0.1",
|
|
snapshot_wallet=SNAPSHOT_DIRECTORY
|
|
/ "monero-stagenet-preprod-20260721T1140Z.wallet",
|
|
snapshot_keys=SNAPSHOT_DIRECTORY
|
|
/ "monero-stagenet-preprod-20260721T1140Z.keys",
|
|
proof_port=39084,
|
|
stagenet=True,
|
|
address_kind="xmr-stagenet",
|
|
proof_directory=pathlib.Path(
|
|
"/run/wowngeon-wallet-keyfile-proof-monero-stagenet"
|
|
),
|
|
),
|
|
"wownero-mainnet": WalletProfile(
|
|
name="wownero-mainnet",
|
|
unit="wownero-wallet-rpc.service",
|
|
user="jw",
|
|
group="jw",
|
|
binary=pathlib.Path("/usr/bin/wownero-wallet-rpc"),
|
|
live_wallet=pathlib.Path("/home/jw/Programs/wow-11.3.0/game"),
|
|
live_rpc_port=34570,
|
|
# This is the audited legacy listener. The proof does not broaden or change it.
|
|
live_bind_address="0.0.0.0",
|
|
snapshot_wallet=SNAPSHOT_DIRECTORY
|
|
/ "wownero-mainnet-preprod-20260721T1140Z.wallet",
|
|
snapshot_keys=SNAPSHOT_DIRECTORY
|
|
/ "wownero-mainnet-preprod-20260721T1140Z.keys",
|
|
proof_port=39570,
|
|
stagenet=False,
|
|
address_kind="wow-mainnet",
|
|
proof_directory=pathlib.Path(
|
|
"/run/wowngeon-wallet-keyfile-proof-wownero-mainnet"
|
|
),
|
|
),
|
|
}
|
|
|
|
|
|
class Refusal(RuntimeError):
|
|
"""A fixed fail-closed condition. Its detail is deliberately never rendered."""
|
|
|
|
|
|
def refuse(code: str) -> None:
|
|
raise Refusal(code)
|
|
|
|
|
|
def harden_process() -> None:
|
|
os.umask(0o077)
|
|
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
|
|
if LIBC.prctl(PR_SET_DUMPABLE, 0, 0, 0, 0) != 0:
|
|
refuse("dump-hardening")
|
|
if LIBC.prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0) != 0:
|
|
refuse("privilege-hardening")
|
|
|
|
|
|
def protect_buffer(value: bytearray) -> None:
|
|
if not value:
|
|
refuse("empty-protected-buffer")
|
|
address = ctypes.addressof(ctypes.c_char.from_buffer(value))
|
|
page_size = os.sysconf("SC_PAGESIZE")
|
|
page_start = address - (address % page_size)
|
|
page_end = (address + len(value) + page_size - 1) // page_size * page_size
|
|
if LIBC.mlock(ctypes.c_void_p(address), ctypes.c_size_t(len(value))) != 0:
|
|
ctypes.memset(ctypes.c_void_p(address), 0, len(value))
|
|
refuse("secret-lock")
|
|
if LIBC.madvise(
|
|
ctypes.c_void_p(page_start),
|
|
ctypes.c_size_t(page_end - page_start),
|
|
ctypes.c_int(MADV_DONTDUMP),
|
|
) != 0:
|
|
ctypes.memset(ctypes.c_void_p(address), 0, len(value))
|
|
LIBC.munlock(ctypes.c_void_p(address), ctypes.c_size_t(len(value)))
|
|
refuse("secret-dontdump")
|
|
|
|
|
|
def wipe_buffer(value: bytearray | None, protected: bool = False) -> None:
|
|
if value is None or len(value) == 0:
|
|
return
|
|
address = ctypes.addressof(ctypes.c_char.from_buffer(value))
|
|
ctypes.memset(ctypes.c_void_p(address), 0, len(value))
|
|
if protected:
|
|
# mlock is page-granular and Python allocations can share a page. Unlocking one wiped
|
|
# allocation could unlock another live protected buffer, so retain the small lock set until
|
|
# this short-lived helper exits.
|
|
pass
|
|
|
|
|
|
def identity(user: str, group: str) -> tuple[int, int]:
|
|
try:
|
|
return pwd.getpwnam(user).pw_uid, grp.getgrnam(group).gr_gid
|
|
except KeyError:
|
|
refuse("identity")
|
|
|
|
|
|
def open_directory_nofollow(path: pathlib.Path) -> int:
|
|
if not path.is_absolute() or pathlib.Path(os.path.normpath(path)) != path:
|
|
refuse("path")
|
|
descriptor = os.open("/", os.O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC)
|
|
try:
|
|
for component in path.parts[1:]:
|
|
following = os.open(
|
|
component,
|
|
os.O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC,
|
|
dir_fd=descriptor,
|
|
)
|
|
os.close(descriptor)
|
|
descriptor = following
|
|
if not stat.S_ISDIR(os.fstat(descriptor).st_mode):
|
|
refuse("directory")
|
|
return descriptor
|
|
except BaseException:
|
|
os.close(descriptor)
|
|
raise
|
|
|
|
|
|
def require_directory(
|
|
path: pathlib.Path, expected_uid: int, expected_gid: int, expected_mode: int
|
|
) -> os.stat_result:
|
|
descriptor = open_directory_nofollow(path)
|
|
try:
|
|
details = os.fstat(descriptor)
|
|
if (
|
|
details.st_uid,
|
|
details.st_gid,
|
|
stat.S_IMODE(details.st_mode),
|
|
) != (expected_uid, expected_gid, expected_mode):
|
|
refuse("directory-metadata")
|
|
return details
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def file_identity(details: os.stat_result) -> tuple[int, ...]:
|
|
return (
|
|
details.st_dev,
|
|
details.st_ino,
|
|
details.st_uid,
|
|
details.st_gid,
|
|
stat.S_IMODE(details.st_mode),
|
|
details.st_nlink,
|
|
details.st_size,
|
|
details.st_mtime_ns,
|
|
details.st_ctime_ns,
|
|
)
|
|
|
|
|
|
def open_secure_file_at(
|
|
directory_fd: int,
|
|
name: str,
|
|
expected_uid: int,
|
|
expected_gid: int,
|
|
expected_mode: int,
|
|
) -> tuple[int, tuple[int, ...]]:
|
|
try:
|
|
descriptor = os.open(
|
|
name, os.O_RDONLY | O_NOFOLLOW | O_CLOEXEC, dir_fd=directory_fd
|
|
)
|
|
except OSError:
|
|
refuse("source-file")
|
|
try:
|
|
details = os.fstat(descriptor)
|
|
if (
|
|
not stat.S_ISREG(details.st_mode)
|
|
or details.st_nlink != 1
|
|
or details.st_size <= 0
|
|
or (
|
|
details.st_uid,
|
|
details.st_gid,
|
|
stat.S_IMODE(details.st_mode),
|
|
)
|
|
!= (expected_uid, expected_gid, expected_mode)
|
|
):
|
|
refuse("source-metadata")
|
|
return descriptor, file_identity(details)
|
|
except BaseException:
|
|
os.close(descriptor)
|
|
raise
|
|
|
|
|
|
def secure_wipe_fd(descriptor: int) -> None:
|
|
details = os.fstat(descriptor)
|
|
if not stat.S_ISREG(details.st_mode) or details.st_nlink != 1:
|
|
refuse("wipe-target")
|
|
os.lseek(descriptor, 0, os.SEEK_SET)
|
|
remaining = details.st_size
|
|
while remaining:
|
|
block = ZERO_BLOCK if remaining >= len(ZERO_BLOCK) else ZERO_BLOCK[:remaining]
|
|
view = memoryview(block)
|
|
while view:
|
|
written = os.write(descriptor, view)
|
|
if written <= 0:
|
|
refuse("wipe-write")
|
|
view = view[written:]
|
|
remaining -= len(block)
|
|
os.ftruncate(descriptor, 0)
|
|
os.fsync(descriptor)
|
|
|
|
|
|
def copy_one_exact(
|
|
source_fd: int,
|
|
expected_source: tuple[int, ...],
|
|
destination_fd: int,
|
|
destination_name: str,
|
|
expected_destination_uid: int,
|
|
expected_destination_gid: int,
|
|
after_copy: Callable[[], None] | None = None,
|
|
) -> None:
|
|
output = os.open(
|
|
destination_name,
|
|
os.O_RDWR | os.O_CREAT | os.O_EXCL | O_NOFOLLOW | O_CLOEXEC,
|
|
0o600,
|
|
dir_fd=destination_fd,
|
|
)
|
|
published = False
|
|
try:
|
|
os.lseek(source_fd, 0, os.SEEK_SET)
|
|
while True:
|
|
block = os.read(source_fd, 1024 * 1024)
|
|
if not block:
|
|
break
|
|
view = memoryview(block)
|
|
while view:
|
|
written = os.write(output, view)
|
|
if written <= 0:
|
|
refuse("copy-write")
|
|
view = view[written:]
|
|
os.fsync(output)
|
|
if after_copy is not None:
|
|
after_copy()
|
|
if file_identity(os.fstat(source_fd)) != expected_source:
|
|
refuse("source-race")
|
|
os.lseek(source_fd, 0, os.SEEK_SET)
|
|
os.lseek(output, 0, os.SEEK_SET)
|
|
while True:
|
|
source_block = os.read(source_fd, 1024 * 1024)
|
|
copy_block = os.read(output, 1024 * 1024)
|
|
if source_block != copy_block:
|
|
refuse("copy-mismatch")
|
|
if not source_block:
|
|
break
|
|
output_details = os.fstat(output)
|
|
if (
|
|
not stat.S_ISREG(output_details.st_mode)
|
|
or output_details.st_uid != expected_destination_uid
|
|
or output_details.st_gid != expected_destination_gid
|
|
or output_details.st_nlink != 1
|
|
or stat.S_IMODE(output_details.st_mode) != 0o600
|
|
):
|
|
refuse("copy-mode")
|
|
published = True
|
|
os.fsync(destination_fd)
|
|
finally:
|
|
if not published:
|
|
try:
|
|
secure_wipe_fd(output)
|
|
except BaseException:
|
|
pass
|
|
os.close(output)
|
|
if not published:
|
|
try:
|
|
os.unlink(destination_name, dir_fd=destination_fd)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def copy_snapshot_pair(
|
|
source_directory: pathlib.Path,
|
|
source_wallet_name: str,
|
|
source_keys_name: str,
|
|
destination_directory: pathlib.Path,
|
|
expected_uid: int,
|
|
expected_gid: int,
|
|
after_first_copy: Callable[[], None] | None = None,
|
|
) -> None:
|
|
source_fd = open_directory_nofollow(source_directory)
|
|
destination_fd = open_directory_nofollow(destination_directory)
|
|
opened: list[tuple[int, tuple[int, ...]]] = []
|
|
try:
|
|
opened.append(
|
|
open_secure_file_at(
|
|
source_fd, source_wallet_name, expected_uid, expected_gid, 0o600
|
|
)
|
|
)
|
|
opened.append(
|
|
open_secure_file_at(
|
|
source_fd, source_keys_name, expected_uid, expected_gid, 0o600
|
|
)
|
|
)
|
|
copy_one_exact(
|
|
opened[0][0],
|
|
opened[0][1],
|
|
destination_fd,
|
|
"wallet",
|
|
expected_uid,
|
|
expected_gid,
|
|
after_first_copy,
|
|
)
|
|
copy_one_exact(
|
|
opened[1][0],
|
|
opened[1][1],
|
|
destination_fd,
|
|
"wallet.keys",
|
|
expected_uid,
|
|
expected_gid,
|
|
)
|
|
for descriptor, expected in opened:
|
|
if file_identity(os.fstat(descriptor)) != expected:
|
|
refuse("pair-source-race")
|
|
finally:
|
|
for descriptor, _ in opened:
|
|
os.close(descriptor)
|
|
os.close(source_fd)
|
|
os.close(destination_fd)
|
|
|
|
|
|
def span_equals(buffer: bytearray, span: tuple[int, int], expected: bytes) -> bool:
|
|
start, end = span
|
|
return end - start == len(expected) and all(
|
|
buffer[start + index] == value for index, value in enumerate(expected)
|
|
)
|
|
|
|
|
|
def span_starts(buffer: bytearray, span: tuple[int, int], expected: bytes) -> bool:
|
|
start, end = span
|
|
return end - start >= len(expected) and all(
|
|
buffer[start + index] == value for index, value in enumerate(expected)
|
|
)
|
|
|
|
|
|
def argv_spans(buffer: bytearray, length: int) -> list[tuple[int, int]]:
|
|
if length < 2 or buffer[length - 1] != 0:
|
|
refuse("cmdline-shape")
|
|
spans: list[tuple[int, int]] = []
|
|
start = 0
|
|
for index in range(length):
|
|
if buffer[index] == 0:
|
|
spans.append((start, index))
|
|
start = index + 1
|
|
if not spans or spans[0][0] == spans[0][1]:
|
|
refuse("cmdline-empty")
|
|
return spans
|
|
|
|
|
|
def option_value_span(
|
|
buffer: bytearray, spans: list[tuple[int, int]], option: bytes
|
|
) -> tuple[int, int]:
|
|
matches: list[tuple[int, int]] = []
|
|
inline_prefix = option + b"="
|
|
for index, span in enumerate(spans):
|
|
if span_equals(buffer, span, option):
|
|
if index + 1 >= len(spans):
|
|
refuse("option-value")
|
|
matches.append(spans[index + 1])
|
|
elif span_starts(buffer, span, inline_prefix):
|
|
matches.append((span[0] + len(inline_prefix), span[1]))
|
|
if len(matches) != 1 or matches[0][0] == matches[0][1]:
|
|
refuse("option-count")
|
|
return matches[0]
|
|
|
|
|
|
def count_arg(buffer: bytearray, spans: list[tuple[int, int]], value: bytes) -> int:
|
|
return sum(1 for span in spans if span_equals(buffer, span, value))
|
|
|
|
|
|
def count_inline_option(
|
|
buffer: bytearray, spans: list[tuple[int, int]], option: bytes
|
|
) -> int:
|
|
prefix = option + b"="
|
|
return sum(1 for span in spans if span_starts(buffer, span, prefix))
|
|
|
|
|
|
def extract_inline_wallet_password(
|
|
buffer: bytearray,
|
|
length: int,
|
|
profile: WalletProfile,
|
|
protect_secret: bool = False,
|
|
buffer_is_protected: bool = False,
|
|
) -> bytearray:
|
|
secret: bytearray | None = None
|
|
try:
|
|
spans = argv_spans(buffer, length)
|
|
if not span_equals(buffer, spans[0], os.fsencode(profile.binary)):
|
|
refuse("argv-binary")
|
|
if any(
|
|
span_equals(buffer, span, forbidden)
|
|
or span_starts(buffer, span, forbidden + b"=")
|
|
for span in spans
|
|
for forbidden in (b"--password-file", b"--wallet-dir")
|
|
):
|
|
refuse("argv-indirect-wallet")
|
|
wallet_span = option_value_span(buffer, spans, b"--wallet-file")
|
|
if not span_equals(buffer, wallet_span, os.fsencode(profile.live_wallet)):
|
|
refuse("argv-wallet")
|
|
port_span = option_value_span(buffer, spans, b"--rpc-bind-port")
|
|
if not span_equals(buffer, port_span, str(profile.live_rpc_port).encode("ascii")):
|
|
refuse("argv-port")
|
|
if profile.stagenet:
|
|
if (
|
|
count_arg(buffer, spans, b"--stagenet") != 1
|
|
or count_inline_option(buffer, spans, b"--stagenet") != 0
|
|
):
|
|
refuse("argv-network")
|
|
elif (
|
|
count_arg(buffer, spans, b"--stagenet") != 0
|
|
or count_inline_option(buffer, spans, b"--stagenet") != 0
|
|
):
|
|
refuse("argv-network")
|
|
if (
|
|
count_arg(buffer, spans, b"--testnet") != 0
|
|
or count_inline_option(buffer, spans, b"--testnet") != 0
|
|
):
|
|
refuse("argv-network")
|
|
password_span = option_value_span(buffer, spans, b"--password")
|
|
password_length = password_span[1] - password_span[0]
|
|
if password_length < 1 or password_length > 4096:
|
|
refuse("password-shape")
|
|
if any(buffer[index] in (10, 13) for index in range(*password_span)):
|
|
refuse("password-shape")
|
|
secret = bytearray(memoryview(buffer)[password_span[0] : password_span[1]])
|
|
if protect_secret:
|
|
protect_buffer(secret)
|
|
return secret
|
|
finally:
|
|
wipe_buffer(buffer, buffer_is_protected)
|
|
|
|
|
|
def read_live_cmdline_password(pid: int, profile: WalletProfile) -> bytearray:
|
|
buffer = bytearray(MAX_CMDLINE)
|
|
protect_buffer(buffer)
|
|
try:
|
|
descriptor = os.open(
|
|
f"/proc/{pid}/cmdline", os.O_RDONLY | O_NOFOLLOW | O_CLOEXEC
|
|
)
|
|
try:
|
|
if not stat.S_ISREG(os.fstat(descriptor).st_mode):
|
|
refuse("cmdline-file")
|
|
length = os.readv(descriptor, [buffer])
|
|
overflow = bytearray(1)
|
|
try:
|
|
if os.readv(descriptor, [overflow]) != 0:
|
|
refuse("cmdline-size")
|
|
finally:
|
|
wipe_buffer(overflow)
|
|
finally:
|
|
os.close(descriptor)
|
|
return extract_inline_wallet_password(
|
|
buffer,
|
|
length,
|
|
profile,
|
|
protect_secret=True,
|
|
buffer_is_protected=True,
|
|
)
|
|
except BaseException:
|
|
# extract_inline_wallet_password already wipes on the normal parsing path.
|
|
wipe_buffer(buffer, True)
|
|
raise
|
|
|
|
|
|
def systemctl(*arguments: str) -> subprocess.CompletedProcess[bytes]:
|
|
try:
|
|
return subprocess.run(
|
|
["/usr/bin/systemctl", *arguments],
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
timeout=20,
|
|
env={"PATH": "/usr/bin:/bin", "LANG": "C", "LC_ALL": "C"},
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
refuse("systemd")
|
|
|
|
|
|
def main_pid(unit: str) -> int:
|
|
result = systemctl("show", unit, "--property=MainPID", "--value")
|
|
if result.returncode != 0:
|
|
refuse("mainpid")
|
|
try:
|
|
return int(result.stdout.strip())
|
|
except ValueError:
|
|
refuse("mainpid")
|
|
|
|
|
|
def process_status_ids(pid: int, field: str) -> tuple[int, int, int, int]:
|
|
try:
|
|
with open(f"/proc/{pid}/status", "r", encoding="ascii") as source:
|
|
for line in source:
|
|
if line.startswith(f"{field}:"):
|
|
values = line.split()[1:]
|
|
if len(values) == 4:
|
|
return (
|
|
int(values[0]),
|
|
int(values[1]),
|
|
int(values[2]),
|
|
int(values[3]),
|
|
)
|
|
except (OSError, UnicodeDecodeError, ValueError):
|
|
pass
|
|
refuse("process-identity")
|
|
|
|
|
|
def process_uid(pid: int) -> int:
|
|
values = process_status_ids(pid, "Uid")
|
|
if len(set(values)) != 1:
|
|
refuse("process-uid")
|
|
return values[0]
|
|
|
|
|
|
def process_gid(pid: int) -> int:
|
|
values = process_status_ids(pid, "Gid")
|
|
if len(set(values)) != 1:
|
|
refuse("process-gid")
|
|
return values[0]
|
|
|
|
|
|
def process_groups(pid: int) -> list[int]:
|
|
try:
|
|
with open(f"/proc/{pid}/status", "r", encoding="ascii") as source:
|
|
for line in source:
|
|
if line.startswith("Groups:"):
|
|
return [int(value) for value in line.split()[1:]]
|
|
except (OSError, UnicodeDecodeError, ValueError):
|
|
pass
|
|
refuse("process-groups")
|
|
|
|
|
|
def process_status_integer(pid: int, field: str) -> int:
|
|
try:
|
|
with open(f"/proc/{pid}/status", "r", encoding="ascii") as source:
|
|
for line in source:
|
|
if line.startswith(f"{field}:"):
|
|
values = line.split()[1:]
|
|
if len(values) == 1:
|
|
return int(values[0])
|
|
except (OSError, UnicodeDecodeError, ValueError):
|
|
pass
|
|
refuse("process-security-state")
|
|
|
|
|
|
def process_status_hex(pid: int, field: str) -> int:
|
|
try:
|
|
with open(f"/proc/{pid}/status", "r", encoding="ascii") as source:
|
|
for line in source:
|
|
if line.startswith(f"{field}:"):
|
|
values = line.split()[1:]
|
|
if len(values) == 1:
|
|
return int(values[0], 16)
|
|
except (OSError, UnicodeDecodeError, ValueError):
|
|
pass
|
|
refuse("process-capabilities")
|
|
|
|
|
|
def process_start_time(pid: int) -> str:
|
|
try:
|
|
raw = pathlib.Path(f"/proc/{pid}/stat").read_text(encoding="ascii")
|
|
fields = raw[raw.rfind(")") + 2 :].split()
|
|
return fields[19]
|
|
except (OSError, UnicodeDecodeError, IndexError):
|
|
refuse("process-start")
|
|
|
|
|
|
def process_has_unit_cgroup(pid: int, unit: str) -> bool:
|
|
try:
|
|
lines = pathlib.Path(f"/proc/{pid}/cgroup").read_text(
|
|
encoding="ascii"
|
|
).splitlines()
|
|
except (OSError, UnicodeDecodeError):
|
|
refuse("process-cgroup")
|
|
return any(line.rsplit(":", 1)[-1].endswith(f"/{unit}") for line in lines)
|
|
|
|
|
|
def process_executable(pid: int) -> pathlib.Path:
|
|
try:
|
|
return pathlib.Path(os.readlink(f"/proc/{pid}/exe"))
|
|
except OSError:
|
|
refuse("process-executable")
|
|
|
|
|
|
def process_executable_identity(pid: int) -> tuple[int, ...]:
|
|
try:
|
|
descriptor = os.open(f"/proc/{pid}/exe", os.O_RDONLY | O_CLOEXEC)
|
|
except OSError:
|
|
refuse("process-executable-file")
|
|
try:
|
|
details = os.fstat(descriptor)
|
|
if not stat.S_ISREG(details.st_mode):
|
|
refuse("process-executable-file")
|
|
return file_identity(details)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def parse_listener_table(
|
|
path: str, port: int, ipv6: bool
|
|
) -> list[tuple[str, int]]:
|
|
found: list[tuple[str, int]] = []
|
|
try:
|
|
with open(path, "r", encoding="ascii") as source:
|
|
next(source, None)
|
|
for line in source:
|
|
fields = line.split()
|
|
if len(fields) < 10 or fields[3] != "0A":
|
|
continue
|
|
address_hex, port_hex = fields[1].split(":")
|
|
if int(port_hex, 16) != port:
|
|
continue
|
|
if ipv6:
|
|
address = socket.inet_ntop(
|
|
socket.AF_INET6, bytes.fromhex(address_hex)
|
|
)
|
|
else:
|
|
address = socket.inet_ntoa(bytes.fromhex(address_hex)[::-1])
|
|
found.append((address, int(fields[9])))
|
|
except (OSError, UnicodeDecodeError, ValueError, IndexError):
|
|
refuse("listener-table")
|
|
return found
|
|
|
|
|
|
def listeners(port: int) -> list[tuple[str, int]]:
|
|
return parse_listener_table("/proc/net/tcp", port, False) + parse_listener_table(
|
|
"/proc/net/tcp6", port, True
|
|
)
|
|
|
|
|
|
def process_owns_socket(pid: int, inode: int) -> bool:
|
|
expected = f"socket:[{inode}]"
|
|
try:
|
|
for entry in os.scandir(f"/proc/{pid}/fd"):
|
|
try:
|
|
if os.readlink(entry.path) == expected:
|
|
return True
|
|
except OSError:
|
|
continue
|
|
except OSError:
|
|
refuse("process-fds")
|
|
return False
|
|
|
|
|
|
def require_exact_listener(pid: int, port: int, bind_address: str) -> None:
|
|
active = listeners(port)
|
|
if len(active) != 1 or active[0][0] != bind_address:
|
|
refuse("listener-identity")
|
|
if not process_owns_socket(pid, active[0][1]):
|
|
refuse("listener-owner")
|
|
|
|
|
|
def require_unused_port(port: int) -> None:
|
|
if listeners(port):
|
|
refuse("proof-port-collision")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LiveIdentity:
|
|
pid: int
|
|
start_time: str
|
|
binary_identity: tuple[int, ...]
|
|
|
|
|
|
def require_live_process(
|
|
profile: WalletProfile, expected_binary: tuple[int, ...]
|
|
) -> LiveIdentity:
|
|
if systemctl("is-active", "--quiet", profile.unit).returncode != 0:
|
|
refuse("unit-inactive")
|
|
pid = main_pid(profile.unit)
|
|
if pid <= 1:
|
|
refuse("unit-pid")
|
|
expected_uid, expected_gid = identity(profile.user, profile.group)
|
|
started = process_start_time(pid)
|
|
if process_uid(pid) != expected_uid:
|
|
refuse("live-uid")
|
|
if process_gid(pid) != expected_gid:
|
|
refuse("live-gid")
|
|
if process_executable(pid) != profile.binary:
|
|
refuse("live-executable")
|
|
if process_executable_identity(pid) != expected_binary:
|
|
refuse("live-executable-identity")
|
|
if not process_has_unit_cgroup(pid, profile.unit):
|
|
refuse("live-cgroup")
|
|
require_exact_listener(
|
|
pid, profile.live_rpc_port, profile.live_bind_address
|
|
)
|
|
return LiveIdentity(
|
|
pid=pid, start_time=started, binary_identity=expected_binary
|
|
)
|
|
|
|
|
|
def require_live_unchanged(profile: WalletProfile, live: LiveIdentity) -> None:
|
|
if main_pid(profile.unit) != live.pid:
|
|
refuse("live-race")
|
|
if process_start_time(live.pid) != live.start_time:
|
|
refuse("live-race")
|
|
expected_uid, expected_gid = identity(profile.user, profile.group)
|
|
if process_uid(live.pid) != expected_uid:
|
|
refuse("live-race")
|
|
if process_gid(live.pid) != expected_gid:
|
|
refuse("live-race")
|
|
if process_executable(live.pid) != profile.binary:
|
|
refuse("live-race")
|
|
if (
|
|
process_executable_identity(live.pid) != live.binary_identity
|
|
or require_binary(profile) != live.binary_identity
|
|
):
|
|
refuse("live-race")
|
|
if not process_has_unit_cgroup(live.pid, profile.unit):
|
|
refuse("live-race")
|
|
require_exact_listener(
|
|
live.pid, profile.live_rpc_port, profile.live_bind_address
|
|
)
|
|
|
|
|
|
def network_shaped_address(value: object, profile: WalletProfile) -> bool:
|
|
if not isinstance(value, str) or not value.isascii():
|
|
return False
|
|
if profile.address_kind == "xmr-stagenet":
|
|
return (
|
|
len(value) == 95
|
|
and value.startswith("5")
|
|
and all(character in BASE58_ALPHABET for character in value)
|
|
)
|
|
if profile.address_kind == "wow-mainnet":
|
|
return (
|
|
len(value) == 97
|
|
and value.startswith("Wo")
|
|
and all(character in BASE58_ALPHABET for character in value)
|
|
)
|
|
return False
|
|
|
|
|
|
def rpc_post(
|
|
port: int,
|
|
request_body: bytes | bytearray = ADDRESS_REQUEST,
|
|
authorization: str | None = None,
|
|
) -> tuple[int, str | None, bytes]:
|
|
connection = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
|
|
headers = {"Content-Type": "application/json", "Accept": "application/json"}
|
|
if authorization is not None:
|
|
headers["Authorization"] = authorization
|
|
try:
|
|
connection.request("POST", RPC_PATH, body=request_body, headers=headers)
|
|
response = connection.getresponse()
|
|
challenge = response.getheader("WWW-Authenticate")
|
|
raw = response.read(MAX_RPC_RESPONSE + 1)
|
|
if len(raw) > MAX_RPC_RESPONSE:
|
|
refuse("rpc-size")
|
|
return response.status, challenge, raw
|
|
except (OSError, TimeoutError, http.client.HTTPException):
|
|
refuse("rpc-transport")
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def address_from_response(raw: bytes, profile: WalletProfile) -> str:
|
|
try:
|
|
payload = json.loads(raw.decode("utf-8"))
|
|
address = payload["result"]["address"]
|
|
except (UnicodeDecodeError, json.JSONDecodeError, KeyError, TypeError):
|
|
refuse("rpc-payload")
|
|
if not network_shaped_address(address, profile):
|
|
refuse("address-network")
|
|
return address
|
|
|
|
|
|
def current_live_address(profile: WalletProfile) -> str:
|
|
status, _, raw = rpc_post(profile.live_rpc_port)
|
|
if status != 200:
|
|
refuse("live-rpc-status")
|
|
return address_from_response(raw, profile)
|
|
|
|
|
|
def parse_digest_challenge(header: str | None) -> dict[str, str]:
|
|
if header is None or len(header) > 4096 or "\r" in header or "\n" in header:
|
|
refuse("digest-challenge")
|
|
scheme, separator, fields = header.partition(" ")
|
|
if scheme.lower() != "digest" or not separator:
|
|
refuse("digest-scheme")
|
|
try:
|
|
parsed = urllib.request.parse_keqv_list(urllib.request.parse_http_list(fields))
|
|
except (TypeError, ValueError):
|
|
refuse("digest-parse")
|
|
if not isinstance(parsed, dict):
|
|
refuse("digest-parse")
|
|
realm = parsed.get("realm")
|
|
nonce = parsed.get("nonce")
|
|
qop = parsed.get("qop")
|
|
algorithm = parsed.get("algorithm", "MD5")
|
|
qop_values = {value.strip().lower() for value in qop.split(",")} if qop else set()
|
|
if not realm or not nonce or "auth" not in qop_values:
|
|
refuse("digest-fields")
|
|
if algorithm.upper() != "MD5":
|
|
refuse("digest-algorithm")
|
|
for value in (realm, nonce, parsed.get("opaque", "")):
|
|
if not value.isascii() or any(character in value for character in ('"', "\\")):
|
|
refuse("digest-value")
|
|
return parsed
|
|
|
|
|
|
def md5_hex(*parts: bytes | bytearray | memoryview) -> str:
|
|
try:
|
|
digest = hashlib.md5(usedforsecurity=False)
|
|
except TypeError: # pragma: no cover - compatibility with older Python/OpenSSL builds
|
|
digest = hashlib.md5()
|
|
for part in parts:
|
|
digest.update(part)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def random_hex(byte_count: int) -> bytearray:
|
|
entropy = bytearray(byte_count)
|
|
protect_buffer(entropy)
|
|
output = bytearray(byte_count * 2)
|
|
try:
|
|
protect_buffer(output)
|
|
descriptor = os.open("/dev/urandom", os.O_RDONLY | O_NOFOLLOW | O_CLOEXEC)
|
|
try:
|
|
if os.readv(descriptor, [entropy]) != byte_count:
|
|
refuse("random-read")
|
|
finally:
|
|
os.close(descriptor)
|
|
alphabet = b"0123456789abcdef"
|
|
for index, value in enumerate(entropy):
|
|
output[index * 2] = alphabet[value >> 4]
|
|
output[index * 2 + 1] = alphabet[value & 15]
|
|
return output
|
|
except BaseException:
|
|
wipe_buffer(output, True)
|
|
raise
|
|
finally:
|
|
wipe_buffer(entropy, True)
|
|
|
|
|
|
def digest_authorization(
|
|
challenge: str, rpc_secret: bytearray, nonce_count: int = 1
|
|
) -> str:
|
|
if nonce_count < 1 or nonce_count > 0xFFFFFFFF:
|
|
refuse("digest-count")
|
|
parsed = parse_digest_challenge(challenge)
|
|
realm = parsed["realm"]
|
|
nonce = parsed["nonce"]
|
|
cnonce = random_hex(16)
|
|
try:
|
|
cnonce_text = bytes(cnonce).decode("ascii")
|
|
ha1 = md5_hex(
|
|
RPC_USER.encode("ascii"),
|
|
b":",
|
|
realm.encode("ascii"),
|
|
b":",
|
|
memoryview(rpc_secret),
|
|
)
|
|
ha2 = md5_hex(b"POST", b":", RPC_PATH.encode("ascii"))
|
|
nonce_count_text = f"{nonce_count:08x}"
|
|
response = md5_hex(
|
|
ha1.encode("ascii"),
|
|
b":",
|
|
nonce.encode("ascii"),
|
|
b":",
|
|
nonce_count_text.encode("ascii"),
|
|
b":",
|
|
cnonce_text.encode("ascii"),
|
|
b":auth:",
|
|
ha2.encode("ascii"),
|
|
)
|
|
fields = [
|
|
f'username="{RPC_USER}"',
|
|
f'realm="{realm}"',
|
|
f'nonce="{nonce}"',
|
|
f'uri="{RPC_PATH}"',
|
|
"algorithm=MD5",
|
|
f'response="{response}"',
|
|
"qop=auth",
|
|
f"nc={nonce_count_text}",
|
|
f'cnonce="{cnonce_text}"',
|
|
]
|
|
if parsed.get("opaque"):
|
|
fields.append(f'opaque="{parsed["opaque"]}"')
|
|
return "Digest " + ", ".join(fields)
|
|
finally:
|
|
wipe_buffer(cnonce, True)
|
|
|
|
|
|
def authenticated_candidate_rpc(
|
|
profile: WalletProfile,
|
|
challenge: str,
|
|
rpc_secret: bytearray,
|
|
request_body: bytes | bytearray,
|
|
nonce_count: int,
|
|
) -> bytes:
|
|
authorization = digest_authorization(challenge, rpc_secret, nonce_count)
|
|
status, _, raw = rpc_post(profile.proof_port, request_body, authorization)
|
|
if status != 200:
|
|
refuse("authenticated-rpc-status")
|
|
return raw
|
|
|
|
|
|
def candidate_address(profile: WalletProfile, challenge: str, rpc_secret: bytearray) -> str:
|
|
raw = authenticated_candidate_rpc(
|
|
profile, challenge, rpc_secret, ADDRESS_REQUEST, 1
|
|
)
|
|
return address_from_response(raw, profile)
|
|
|
|
|
|
def rpc_request(method: str, params: dict[str, object]) -> bytearray:
|
|
if method not in {"sign", "verify"}:
|
|
refuse("rpc-method")
|
|
request = bytearray(
|
|
json.dumps(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": "keyfile-recovery-proof",
|
|
"method": method,
|
|
"params": params,
|
|
},
|
|
separators=(",", ":"),
|
|
).encode("ascii")
|
|
)
|
|
protect_buffer(request)
|
|
return request
|
|
|
|
|
|
def prove_spend_capability(
|
|
profile: WalletProfile,
|
|
challenge: str,
|
|
rpc_secret: bytearray,
|
|
restored_address: str,
|
|
) -> None:
|
|
sign_request = rpc_request("sign", {"data": SIGN_MESSAGE})
|
|
try:
|
|
sign_raw = authenticated_candidate_rpc(
|
|
profile, challenge, rpc_secret, sign_request, 2
|
|
)
|
|
finally:
|
|
wipe_buffer(sign_request, True)
|
|
try:
|
|
sign_payload = json.loads(sign_raw.decode("utf-8"))
|
|
signature = sign_payload["result"]["signature"]
|
|
except (UnicodeDecodeError, json.JSONDecodeError, KeyError, TypeError):
|
|
refuse("sign-payload")
|
|
if (
|
|
not isinstance(signature, str)
|
|
or not signature.isascii()
|
|
or len(signature) < 16
|
|
or len(signature) > 4096
|
|
or not signature.startswith("Sig")
|
|
or any(character in ('"', "\\", "\r", "\n") for character in signature)
|
|
):
|
|
refuse("signature-shape")
|
|
verify_request = rpc_request(
|
|
"verify",
|
|
{
|
|
"data": SIGN_MESSAGE,
|
|
"address": restored_address,
|
|
"signature": signature,
|
|
},
|
|
)
|
|
try:
|
|
verify_raw = authenticated_candidate_rpc(
|
|
profile, challenge, rpc_secret, verify_request, 3
|
|
)
|
|
finally:
|
|
wipe_buffer(verify_request, True)
|
|
try:
|
|
verify_result = json.loads(verify_raw.decode("utf-8"))["result"]
|
|
good = verify_result["good"]
|
|
signature_type = verify_result.get("signature_type")
|
|
except (UnicodeDecodeError, json.JSONDecodeError, KeyError, TypeError):
|
|
refuse("verify-payload")
|
|
if good is not True or signature_type not in (None, "spend"):
|
|
refuse("spend-capability")
|
|
|
|
|
|
def wait_for_digest_challenge(
|
|
profile: WalletProfile, process: subprocess.Popen[bytes]
|
|
) -> str:
|
|
deadline = time.monotonic() + 25
|
|
while time.monotonic() < deadline:
|
|
if process.poll() is not None:
|
|
refuse("candidate-exited")
|
|
try:
|
|
status, challenge, _ = rpc_post(profile.proof_port, ADDRESS_REQUEST)
|
|
if status != 401:
|
|
refuse("unauthenticated-not-refused")
|
|
parse_digest_challenge(challenge)
|
|
require_exact_listener(
|
|
process.pid, profile.proof_port, "127.0.0.1"
|
|
)
|
|
return challenge or ""
|
|
except Refusal as problem:
|
|
if str(problem) not in {"rpc-transport", "listener-identity"}:
|
|
raise
|
|
time.sleep(0.1)
|
|
refuse("candidate-timeout")
|
|
|
|
|
|
def write_new_file(directory_fd: int, name: str, value: bytearray) -> None:
|
|
descriptor = os.open(
|
|
name,
|
|
os.O_WRONLY | os.O_CREAT | os.O_EXCL | O_NOFOLLOW | O_CLOEXEC,
|
|
0o600,
|
|
dir_fd=directory_fd,
|
|
)
|
|
try:
|
|
view = memoryview(value)
|
|
while view:
|
|
written = os.write(descriptor, view)
|
|
if written <= 0:
|
|
refuse("secret-write")
|
|
view = view[written:]
|
|
os.fsync(descriptor)
|
|
details = os.fstat(descriptor)
|
|
if (
|
|
details.st_uid != 0
|
|
or details.st_gid != 0
|
|
or details.st_nlink != 1
|
|
or stat.S_IMODE(details.st_mode) != 0o600
|
|
):
|
|
refuse("secret-file-metadata")
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def prepare_candidate_files(
|
|
profile: WalletProfile, wallet_password: bytearray
|
|
) -> bytearray:
|
|
rpc_secret = random_hex(32)
|
|
directory_fd = open_directory_nofollow(profile.candidate_directory)
|
|
password_file = bytearray(wallet_password)
|
|
password_file.extend(b"\n")
|
|
config = bytearray()
|
|
try:
|
|
protect_buffer(password_file)
|
|
write_new_file(directory_fd, "wallet-password", password_file)
|
|
config.extend(
|
|
f"wallet-file={profile.candidate_directory}/wallet\n"
|
|
f"password-file={profile.candidate_directory}/wallet-password\n"
|
|
f"rpc-bind-port={profile.proof_port}\n"
|
|
"rpc-bind-ip=127.0.0.1\n"
|
|
"non-interactive=1\n"
|
|
f"rpc-login={RPC_USER}:".encode("ascii")
|
|
)
|
|
config.extend(rpc_secret)
|
|
config.extend(b"\n")
|
|
if profile.stagenet:
|
|
config.extend(b"stagenet=1\n")
|
|
protect_buffer(config)
|
|
try:
|
|
write_new_file(directory_fd, "wallet-rpc.conf", config)
|
|
finally:
|
|
wipe_buffer(config, True)
|
|
os.fsync(directory_fd)
|
|
return rpc_secret
|
|
except BaseException:
|
|
wipe_buffer(rpc_secret, True)
|
|
raise
|
|
finally:
|
|
wipe_buffer(password_file, True)
|
|
os.close(directory_fd)
|
|
|
|
|
|
def require_binary(profile: WalletProfile) -> tuple[int, ...]:
|
|
parent_fd = open_directory_nofollow(profile.binary.parent)
|
|
try:
|
|
descriptor, expected = open_secure_file_at(
|
|
parent_fd, profile.binary.name, 0, 0, 0o755
|
|
)
|
|
os.close(descriptor)
|
|
return expected
|
|
finally:
|
|
os.close(parent_fd)
|
|
|
|
|
|
def require_candidate_process(
|
|
profile: WalletProfile,
|
|
process: subprocess.Popen[bytes],
|
|
expected_binary: tuple[int, ...],
|
|
expected_uid: int,
|
|
expected_gid: int,
|
|
) -> None:
|
|
if process.poll() is not None:
|
|
refuse("candidate-exited")
|
|
if process_uid(process.pid) != expected_uid:
|
|
refuse("candidate-uid")
|
|
if process_gid(process.pid) != expected_gid:
|
|
refuse("candidate-gid")
|
|
if process_groups(process.pid):
|
|
refuse("candidate-groups")
|
|
if process_status_integer(process.pid, "NoNewPrivs") != 1:
|
|
refuse("candidate-no-new-privileges")
|
|
if any(
|
|
process_status_hex(process.pid, field) != 0
|
|
for field in ("CapInh", "CapPrm", "CapEff", "CapAmb")
|
|
):
|
|
refuse("candidate-capabilities")
|
|
if process_executable(process.pid) != profile.binary:
|
|
refuse("candidate-executable")
|
|
if process_executable_identity(process.pid) != expected_binary:
|
|
refuse("candidate-executable-identity")
|
|
if require_binary(profile) != expected_binary:
|
|
refuse("candidate-binary-race")
|
|
|
|
|
|
def child_hardening(expected_uid: int, expected_gid: int) -> None:
|
|
try:
|
|
if LIBC.prctl(PR_SET_KEEPCAPS, 0, 0, 0, 0) != 0:
|
|
os._exit(126)
|
|
os.setgroups([])
|
|
os.setgid(expected_gid)
|
|
os.setuid(expected_uid)
|
|
if (
|
|
os.getresuid() != (expected_uid, expected_uid, expected_uid)
|
|
or os.getresgid() != (expected_gid, expected_gid, expected_gid)
|
|
or os.getgroups()
|
|
):
|
|
os._exit(126)
|
|
os.umask(0o077)
|
|
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
|
|
# Linux resets dumpability across a credential transition, so set this after setuid.
|
|
if LIBC.prctl(PR_SET_DUMPABLE, 0, 0, 0, 0) != 0:
|
|
os._exit(126)
|
|
if LIBC.prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0) != 0:
|
|
os._exit(126)
|
|
except BaseException:
|
|
os._exit(126)
|
|
|
|
|
|
def launch_candidate(
|
|
profile: WalletProfile, expected_uid: int, expected_gid: int
|
|
) -> subprocess.Popen[bytes]:
|
|
command = [
|
|
os.fspath(profile.binary),
|
|
"--config-file",
|
|
os.fspath(profile.candidate_directory / "wallet-rpc.conf"),
|
|
"--offline",
|
|
"--log-file",
|
|
"/dev/null",
|
|
]
|
|
try:
|
|
return subprocess.Popen(
|
|
command,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
cwd=profile.candidate_directory,
|
|
env={
|
|
"PATH": "/usr/bin:/bin",
|
|
"HOME": os.fspath(profile.candidate_directory),
|
|
"LANG": "C",
|
|
"LC_ALL": "C",
|
|
},
|
|
close_fds=True,
|
|
start_new_session=True,
|
|
preexec_fn=lambda: child_hardening(expected_uid, expected_gid),
|
|
)
|
|
except OSError:
|
|
refuse("candidate-launch")
|
|
|
|
|
|
def stop_candidate(
|
|
profile: WalletProfile, process: subprocess.Popen[bytes] | None
|
|
) -> bool:
|
|
if process is None:
|
|
return not listeners(profile.proof_port)
|
|
clean = True
|
|
if process.poll() is None:
|
|
try:
|
|
process.terminate()
|
|
process.wait(timeout=20)
|
|
except subprocess.TimeoutExpired:
|
|
clean = False
|
|
try:
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
clean = False
|
|
except OSError:
|
|
clean = False
|
|
if process.returncode not in (0, -signal.SIGTERM):
|
|
clean = False
|
|
deadline = time.monotonic() + 3
|
|
while time.monotonic() < deadline and listeners(profile.proof_port):
|
|
time.sleep(0.05)
|
|
return clean and not listeners(profile.proof_port)
|
|
|
|
|
|
def secure_delete_file_at(directory_fd: int, name: str) -> bool:
|
|
try:
|
|
before = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
|
|
if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1:
|
|
os.unlink(name, dir_fd=directory_fd)
|
|
return False
|
|
descriptor = os.open(
|
|
name, os.O_RDWR | O_NOFOLLOW | O_CLOEXEC, dir_fd=directory_fd
|
|
)
|
|
try:
|
|
after = os.fstat(descriptor)
|
|
if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
|
|
return False
|
|
secure_wipe_fd(descriptor)
|
|
finally:
|
|
os.close(descriptor)
|
|
os.unlink(name, dir_fd=directory_fd)
|
|
os.fsync(directory_fd)
|
|
return True
|
|
except FileNotFoundError:
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def secure_clean_directory_fd(directory_fd: int, depth: int = 0) -> bool:
|
|
if depth > 8:
|
|
return False
|
|
clean = True
|
|
try:
|
|
names = os.listdir(directory_fd)
|
|
except OSError:
|
|
return False
|
|
for name in names:
|
|
try:
|
|
details = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
|
|
if stat.S_ISREG(details.st_mode):
|
|
clean = secure_delete_file_at(directory_fd, name) and clean
|
|
elif stat.S_ISDIR(details.st_mode):
|
|
child_fd = os.open(
|
|
name,
|
|
os.O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC,
|
|
dir_fd=directory_fd,
|
|
)
|
|
try:
|
|
clean = secure_clean_directory_fd(child_fd, depth + 1) and clean
|
|
finally:
|
|
os.close(child_fd)
|
|
os.rmdir(name, dir_fd=directory_fd)
|
|
else:
|
|
os.unlink(name, dir_fd=directory_fd)
|
|
clean = False
|
|
except OSError:
|
|
clean = False
|
|
try:
|
|
os.fsync(directory_fd)
|
|
except OSError:
|
|
clean = False
|
|
return clean
|
|
|
|
|
|
def secure_remove_proof_directory(
|
|
path: pathlib.Path, expected_identity: tuple[int, int]
|
|
) -> bool:
|
|
try:
|
|
directory_fd = open_directory_nofollow(path)
|
|
except BaseException:
|
|
return False
|
|
clean = True
|
|
try:
|
|
details = os.fstat(directory_fd)
|
|
if (details.st_dev, details.st_ino) != expected_identity:
|
|
return False
|
|
clean = secure_clean_directory_fd(directory_fd) and clean
|
|
finally:
|
|
os.close(directory_fd)
|
|
try:
|
|
parent_fd = open_directory_nofollow(path.parent)
|
|
try:
|
|
current = os.stat(path.name, dir_fd=parent_fd, follow_symlinks=False)
|
|
if (current.st_dev, current.st_ino) != expected_identity:
|
|
return False
|
|
os.rmdir(path.name, dir_fd=parent_fd)
|
|
os.fsync(parent_fd)
|
|
finally:
|
|
os.close(parent_fd)
|
|
except BaseException:
|
|
return False
|
|
return clean
|
|
|
|
|
|
def create_proof_directory(
|
|
profile: WalletProfile, expected_gid: int
|
|
) -> tuple[int, int]:
|
|
require_directory(pathlib.Path("/run"), 0, 0, 0o755)
|
|
parent_fd = open_directory_nofollow(profile.proof_directory.parent)
|
|
created_identity: tuple[int, int] | None = None
|
|
proof_fd: int | None = None
|
|
try:
|
|
try:
|
|
os.mkdir(profile.proof_directory.name, 0o700, dir_fd=parent_fd)
|
|
except FileExistsError:
|
|
refuse("stale-proof-directory")
|
|
proof_fd = os.open(
|
|
profile.proof_directory.name,
|
|
os.O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC,
|
|
dir_fd=parent_fd,
|
|
)
|
|
details = os.fstat(proof_fd)
|
|
created_identity = details.st_dev, details.st_ino
|
|
os.fchown(proof_fd, 0, expected_gid)
|
|
os.fchmod(proof_fd, 0o710)
|
|
details = os.fstat(proof_fd)
|
|
if (
|
|
not stat.S_ISDIR(details.st_mode)
|
|
or details.st_uid != 0
|
|
or details.st_gid != expected_gid
|
|
or stat.S_IMODE(details.st_mode) != 0o710
|
|
):
|
|
refuse("proof-directory-metadata")
|
|
os.mkdir("candidate", 0o700, dir_fd=proof_fd)
|
|
candidate_fd = os.open(
|
|
"candidate",
|
|
os.O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC,
|
|
dir_fd=proof_fd,
|
|
)
|
|
try:
|
|
os.fchown(candidate_fd, 0, 0)
|
|
os.fchmod(candidate_fd, 0o700)
|
|
candidate_details = os.fstat(candidate_fd)
|
|
if (
|
|
not stat.S_ISDIR(candidate_details.st_mode)
|
|
or candidate_details.st_uid != 0
|
|
or candidate_details.st_gid != 0
|
|
or stat.S_IMODE(candidate_details.st_mode) != 0o700
|
|
):
|
|
refuse("candidate-directory-metadata")
|
|
os.fsync(candidate_fd)
|
|
finally:
|
|
os.close(candidate_fd)
|
|
os.fsync(proof_fd)
|
|
os.fsync(parent_fd)
|
|
return created_identity
|
|
except BaseException:
|
|
if created_identity is not None:
|
|
try:
|
|
current = os.stat(
|
|
profile.proof_directory.name,
|
|
dir_fd=parent_fd,
|
|
follow_symlinks=False,
|
|
)
|
|
if (current.st_dev, current.st_ino) == created_identity:
|
|
if proof_fd is not None:
|
|
try:
|
|
os.rmdir("candidate", dir_fd=proof_fd)
|
|
except OSError:
|
|
pass
|
|
os.rmdir(profile.proof_directory.name, dir_fd=parent_fd)
|
|
os.fsync(parent_fd)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
finally:
|
|
if proof_fd is not None:
|
|
os.close(proof_fd)
|
|
os.close(parent_fd)
|
|
|
|
|
|
CANDIDATE_FILES = ("wallet", "wallet.keys", "wallet-password", "wallet-rpc.conf")
|
|
|
|
|
|
def publish_candidate_tree(
|
|
profile: WalletProfile, expected_uid: int, expected_gid: int
|
|
) -> None:
|
|
require_directory(profile.proof_directory, 0, expected_gid, 0o710)
|
|
directory_fd = open_directory_nofollow(profile.candidate_directory)
|
|
try:
|
|
directory_details = os.fstat(directory_fd)
|
|
if (
|
|
directory_details.st_uid != 0
|
|
or directory_details.st_gid != 0
|
|
or stat.S_IMODE(directory_details.st_mode) != 0o700
|
|
):
|
|
refuse("candidate-directory-prepublish")
|
|
if sorted(os.listdir(directory_fd)) != sorted(CANDIDATE_FILES):
|
|
refuse("candidate-files")
|
|
for name in CANDIDATE_FILES:
|
|
descriptor, _ = open_secure_file_at(directory_fd, name, 0, 0, 0o600)
|
|
try:
|
|
os.fchown(descriptor, expected_uid, expected_gid)
|
|
details = os.fstat(descriptor)
|
|
if (
|
|
details.st_uid != expected_uid
|
|
or details.st_gid != expected_gid
|
|
or details.st_nlink != 1
|
|
or stat.S_IMODE(details.st_mode) != 0o600
|
|
):
|
|
refuse("candidate-file-publish")
|
|
finally:
|
|
os.close(descriptor)
|
|
os.fchown(directory_fd, expected_uid, expected_gid)
|
|
os.fchmod(directory_fd, 0o700)
|
|
os.fsync(directory_fd)
|
|
finally:
|
|
os.close(directory_fd)
|
|
|
|
|
|
def require_published_candidate_tree(
|
|
profile: WalletProfile, expected_uid: int, expected_gid: int
|
|
) -> None:
|
|
require_directory(profile.proof_directory, 0, expected_gid, 0o710)
|
|
require_directory(
|
|
profile.candidate_directory, expected_uid, expected_gid, 0o700
|
|
)
|
|
directory_fd = open_directory_nofollow(profile.candidate_directory)
|
|
try:
|
|
if sorted(os.listdir(directory_fd)) != sorted(CANDIDATE_FILES):
|
|
refuse("candidate-files-race")
|
|
for name in CANDIDATE_FILES:
|
|
descriptor, _ = open_secure_file_at(
|
|
directory_fd, name, expected_uid, expected_gid, 0o600
|
|
)
|
|
os.close(descriptor)
|
|
finally:
|
|
os.close(directory_fd)
|
|
|
|
|
|
def remove_loaded_secret_files(profile: WalletProfile) -> bool:
|
|
directory_fd = open_directory_nofollow(profile.candidate_directory)
|
|
try:
|
|
password_removed = secure_delete_file_at(directory_fd, "wallet-password")
|
|
config_removed = secure_delete_file_at(directory_fd, "wallet-rpc.conf")
|
|
return password_removed and config_removed
|
|
finally:
|
|
os.close(directory_fd)
|
|
|
|
|
|
def run_proof(profile: WalletProfile) -> None:
|
|
if os.getuid() != 0 or os.geteuid() != 0:
|
|
refuse("root-only")
|
|
expected_uid, expected_gid = identity(profile.user, profile.group)
|
|
if expected_uid <= 0 or expected_gid <= 0:
|
|
refuse("non-root-candidate-identity")
|
|
require_unused_port(profile.proof_port)
|
|
expected_binary = require_binary(profile)
|
|
live = require_live_process(profile, expected_binary)
|
|
|
|
candidate: subprocess.Popen[bytes] | None = None
|
|
wallet_password: bytearray | None = None
|
|
rpc_secret: bytearray | None = None
|
|
proof_identity: tuple[int, int] | None = None
|
|
failed = False
|
|
cleanup_ok = True
|
|
try:
|
|
wallet_password = read_live_cmdline_password(live.pid, profile)
|
|
require_live_unchanged(profile, live)
|
|
live_address = current_live_address(profile)
|
|
require_live_unchanged(profile, live)
|
|
blocked_signals = {signal.SIGINT, signal.SIGTERM, signal.SIGHUP, signal.SIGQUIT}
|
|
previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, blocked_signals)
|
|
try:
|
|
proof_identity = create_proof_directory(profile, expected_gid)
|
|
finally:
|
|
signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
|
|
require_directory(
|
|
pathlib.Path("/var/backups/wowngeon"),
|
|
0,
|
|
grp.getgrnam("postgres").gr_gid,
|
|
0o710,
|
|
)
|
|
require_directory(SNAPSHOT_DIRECTORY, 0, 0, 0o700)
|
|
copy_snapshot_pair(
|
|
SNAPSHOT_DIRECTORY,
|
|
profile.snapshot_wallet.name,
|
|
profile.snapshot_keys.name,
|
|
profile.candidate_directory,
|
|
0,
|
|
0,
|
|
)
|
|
rpc_secret = prepare_candidate_files(profile, wallet_password)
|
|
wipe_buffer(wallet_password, True)
|
|
wallet_password = None
|
|
publish_candidate_tree(profile, expected_uid, expected_gid)
|
|
require_published_candidate_tree(profile, expected_uid, expected_gid)
|
|
candidate = launch_candidate(profile, expected_uid, expected_gid)
|
|
require_candidate_process(
|
|
profile, candidate, expected_binary, expected_uid, expected_gid
|
|
)
|
|
challenge = wait_for_digest_challenge(profile, candidate)
|
|
require_candidate_process(
|
|
profile, candidate, expected_binary, expected_uid, expected_gid
|
|
)
|
|
if not remove_loaded_secret_files(profile):
|
|
refuse("loaded-secret-cleanup")
|
|
restored_address = candidate_address(profile, challenge, rpc_secret)
|
|
if not hmac.compare_digest(live_address, restored_address):
|
|
refuse("address-mismatch")
|
|
prove_spend_capability(
|
|
profile, challenge, rpc_secret, restored_address
|
|
)
|
|
wipe_buffer(rpc_secret, True)
|
|
rpc_secret = None
|
|
require_live_unchanged(profile, live)
|
|
require_candidate_process(
|
|
profile, candidate, expected_binary, expected_uid, expected_gid
|
|
)
|
|
require_exact_listener(candidate.pid, profile.proof_port, "127.0.0.1")
|
|
except BaseException:
|
|
failed = True
|
|
finally:
|
|
cleanup_signals = {signal.SIGINT, signal.SIGTERM, signal.SIGHUP, signal.SIGQUIT}
|
|
cleanup_mask: set[signal.Signals] | None = None
|
|
try:
|
|
try:
|
|
cleanup_mask = signal.pthread_sigmask(signal.SIG_BLOCK, cleanup_signals)
|
|
except BaseException:
|
|
cleanup_ok = False
|
|
try:
|
|
wipe_buffer(wallet_password, wallet_password is not None)
|
|
wipe_buffer(rpc_secret, rpc_secret is not None)
|
|
except BaseException:
|
|
cleanup_ok = False
|
|
try:
|
|
cleanup_ok = stop_candidate(profile, candidate) and cleanup_ok
|
|
except BaseException:
|
|
cleanup_ok = False
|
|
if proof_identity is not None:
|
|
try:
|
|
cleanup_ok = (
|
|
secure_remove_proof_directory(
|
|
profile.proof_directory, proof_identity
|
|
)
|
|
and cleanup_ok
|
|
)
|
|
except BaseException:
|
|
cleanup_ok = False
|
|
finally:
|
|
if cleanup_mask is not None:
|
|
signal.pthread_sigmask(signal.SIG_SETMASK, cleanup_mask)
|
|
if failed or not cleanup_ok:
|
|
refuse("proof-failed")
|
|
|
|
|
|
def interrupted(_signum: int, _frame: object) -> None:
|
|
refuse("interrupted")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
arguments = sys.argv[1:] if argv is None else argv
|
|
harden_process()
|
|
for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP, signal.SIGQUIT):
|
|
signal.signal(signum, interrupted)
|
|
if len(arguments) != 1 or arguments[0] not in PROFILES:
|
|
refuse("arguments")
|
|
run_proof(PROFILES[arguments[0]])
|
|
return 0
|
|
|
|
|
|
def emit_status(passed: bool) -> None:
|
|
try:
|
|
os.write(1, b"result=passed\n" if passed else b"result=refused\n")
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
exit_code = 1
|
|
try:
|
|
exit_code = main()
|
|
except BaseException:
|
|
# Never render exception text, a traceback, argv, an address, or secret material.
|
|
exit_code = 1
|
|
emit_status(exit_code == 0)
|
|
raise SystemExit(exit_code)
|