personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Pairing config readers."""
5
6from __future__ import annotations
7
8import socket
9from typing import Any
10
11from think.service import DEFAULT_SERVICE_PORT
12from think.utils import get_config, read_service_port
13
14DEFAULT_TOKEN_TTL_SECONDS = 600
15MIN_TOKEN_TTL_SECONDS = 60
16MAX_TOKEN_TTL_SECONDS = 3600
17
18
19def _pairing_config() -> dict[str, Any]:
20 config = get_config()
21 pairing = config.get("pairing")
22 return pairing if isinstance(pairing, dict) else {}
23
24
25def _clean_str(value: Any) -> str | None:
26 if not isinstance(value, str):
27 return None
28 cleaned = value.strip()
29 return cleaned or None
30
31
32def _detect_lan_ipv4() -> str | None:
33 """Return this host's outward-facing IPv4, or None on failure.
34
35 Uses a UDP socket connect to a routable address so the kernel resolves the
36 outbound interface without sending any packets.
37 """
38
39 try:
40 with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
41 sock.connect(("8.8.8.8", 80))
42 return sock.getsockname()[0]
43 except OSError:
44 return None
45
46
47def get_host_url() -> str:
48 configured = _clean_str(_pairing_config().get("host_url"))
49 if configured is not None:
50 return configured
51 convey_config = get_config().get("convey", {})
52 if convey_config.get("allow_network_access", False):
53 lan_ipv4 = _detect_lan_ipv4()
54 if lan_ipv4 is not None:
55 convey_port = read_service_port("convey") or DEFAULT_SERVICE_PORT
56 return f"http://{lan_ipv4}:{convey_port}"
57 convey_port = read_service_port("convey") or DEFAULT_SERVICE_PORT
58 return f"http://localhost:{convey_port}"
59
60
61def get_token_ttl_seconds() -> int:
62 configured = _pairing_config().get("token_ttl_seconds")
63 try:
64 ttl_seconds = int(configured)
65 except (TypeError, ValueError):
66 ttl_seconds = DEFAULT_TOKEN_TTL_SECONDS
67 return max(MIN_TOKEN_TTL_SECONDS, min(MAX_TOKEN_TTL_SECONDS, ttl_seconds))
68
69
70def get_owner_identity() -> str:
71 config = get_config()
72 identity = config.get("identity")
73 if not isinstance(identity, dict):
74 return ""
75 preferred = _clean_str(identity.get("preferred"))
76 if preferred is not None:
77 return preferred
78 name = _clean_str(identity.get("name"))
79 return name or ""
80
81
82__all__ = [
83 "DEFAULT_TOKEN_TTL_SECONDS",
84 "MAX_TOKEN_TTL_SECONDS",
85 "MIN_TOKEN_TTL_SECONDS",
86 "_detect_lan_ipv4",
87 "get_host_url",
88 "get_owner_identity",
89 "get_token_ttl_seconds",
90]