presence_service.py
Fresh, pinned presence proofs behind an owner-only Unix socket.
Overview
Fresh, pinned presence proofs behind an owner-only Unix socket.
depends on presence_git.py presence_verify.py · used by presence-enroll presence_client.py presenced
API
Cclass ServiceError(RuntimeError)
Configuration or startup failure that keeps the daemon fail-closed.
PresenceService.__init__, _prepare_socket_path, _secure_directory, connect_engine, enroll_device, select_enrollmentCclass Enrollment
One pinned device point and the only credential it may prove.
enroll_device, select_enrollmentFselect_enrollment(path: str, key_id: str | None=None) -> Enrollment
Load exactly one enrollment, or select one explicitly by key id.
daemon_main · calls Enrollment, ServiceErrorF_secure_directory(path: str)
Create a directory with 0o700 permissions (owner-only), validating that it is owned by the current user. Raises ServiceError if the directory exists with different ownership or permissions.
_prepare_socket_path, enroll_device · calls ServiceErrorFenroll_device(port: str, path: str=DEFAULT_ENROLLED, replace: bool=False)
Pin the attached device and credential into one owner-only local file.
enroll_main · calls Enrollment, PresenceEngine.close, ServiceError, _secure_directoryCclass PresenceEngine
Own the serial device and serialize fresh challenge/proof transactions.
connect_engineFPresenceEngine.__init__(self, serial, point: bytes, cred_id: bytes, openssl: str='openssl')
Initialize a presence proof engine with a serial port, curve point, and credential ID. Stores the serial port, point, credential ID, and openssl binary name for use in later prove operations.
PresenceUnixServer.__init__FPresenceEngine.close(self)
Close the serial port held by this presence engine.
_prepare_socket_path, connect_engine, daemon_main, enroll_deviceFPresenceEngine.prove(self, max_cm: int) -> dict
Mint one nonce, acquire one fresh proof, and return a safe verdict.
PresenceService.handleFconnect_engine(port: str, enrolled: Enrollment, openssl: str='openssl') -> PresenceEngine
Open the device once and pin both identities before serving requests.
daemon_main · calls PresenceEngine, PresenceEngine.close, ServiceErrorCclass PresenceService
Validate socket requests before allowing them to touch the serial device.
daemon_mainFPresenceService.__init__(self, engine: PresenceEngine, max_cm: int=40)
Initialize a presence service with an engine and maximum distance policy. Validates that max_cm is a positive integer, raising ServiceError otherwise.
ServiceErrorFPresenceService.handle(self, request) -> dict
Handle an incoming presence proof request. Validates the request contains only op and max_cm, that op is prove, and that the requested distance is a positive integer not exceeding the daemon's policy. Returns ok=true with the proof on success or ok=false with a code and reason on validation failure.
_RequestHandler.handle · calls PresenceEngine.proveCclass _RequestHandler(socketserver.StreamRequestHandler)
Socket request handler for presence proof requests. Reads one JSON request line with timeout, parses it, invokes the presence service, and writes the response as JSON back to the client. Silently handles broken pipes and malformed input by responding with structured error codes rather than closing the connection.
F_RequestHandler._send(self, response)
Send a JSON response line to the client. Serializes the response dict to JSON, appends a newline, and writes it to the socket. Silently handles broken pipe and connection reset errors so the server does not crash on client disconnect.
_RequestHandler.handleF_RequestHandler.handle(self)
Parse one incoming JSON request line, validate it contains op and max_cm fields, dispatch to PresenceService.handle, and send a JSON response line to the client; silently handles client disconnect.
PresenceService.handle, _RequestHandler._sendF_prepare_socket_path(path: str)
Prepare a Unix socket path by securing its parent directory, validating the path is either absent or is a stale socket owned by the current user, and unlinking stale sockets; raises ServiceError if the socket is active or unowned.
PresenceUnixServer.__init__ · calls PresenceEngine.close, ServiceError, _secure_directoryCclass PresenceUnixServer(socketserver.UnixStreamServer)
Single-worker Unix server; every serial proof is inherently serialized.
daemon_mainFPresenceUnixServer.__init__(self, path: str, service: PresenceService, request_timeout: float=DEFAULT_REQUEST_TIMEOUT_S)
Initialize a Unix socket presence server: prepares the socket path, stores service reference and request timeout, and locks down socket permissions to 0o600. Raises ServiceError if socket path is active or unowned by the current user.
PresenceEngine.__init__, _prepare_socket_pathFPresenceUnixServer.close_and_unlink(self)
Close the Unix socket server and unlink the socket file, but only if the file at the expected path is still the same socket (by device and inode). Returns without unlinking if the file is no longer present or has been replaced.
daemon_mainFpositive_cm(value: str) -> int
Parse and validate a distance threshold from a string argument. Returns the integer value if it is between 1 and 65534 inclusive, raising ArgumentTypeError otherwise.
Fbuild_daemon_parser()
Build the argument parser for the presenced daemon. Defines options for serial port, socket path, enrollment file, key selection, maximum distance policy, and openssl binary path.
daemon_mainFbuild_enroll_parser()
Build the argument parser for the presence enrollment tool. Defines options for serial port, output enrollment file, and a flag to replace an existing enrollment.
enroll_mainFenroll_main(argv=None) -> int
Main entry point for the presence enrollment tool: parse arguments and enroll the connected device by pinning it to an owner-only file.
build_enroll_parser, enroll_deviceFdaemon_main(argv=None) -> int
Main entry point for the presenced daemon: parse arguments, connect to the device, start the Unix socket server, and handle SIGINT/SIGTERM by raising KeyboardInterrupt for clean shutdown.
PresenceEngine.close, PresenceService, PresenceUnixServer, PresenceUnixServer.close_and_unlink, build_daemon_parser, connect_engine, select_enrollmentFstop(_signum, _frame)
Signal handler that converts a SIGTERM or SIGINT into a KeyboardInterrupt to shut down the daemon cleanly.