diff --git a/src/dhcpsrv/__init__.py b/src/dhcpsrv/__init__.py index 77a47d9..86f6024 100644 --- a/src/dhcpsrv/__init__.py +++ b/src/dhcpsrv/__init__.py @@ -5,5 +5,5 @@ The single source of truth for the project version. Bump this before tagging a release; CI reads the tag, the code reads this constant. """ -__version__ = "1.2.3" +__version__ = "1.2.4" REPO = "engel/dhcpsrv" # self-hosted Gitea (git.engelgardt23.ru) diff --git a/src/dhcpsrv/app.py b/src/dhcpsrv/app.py index d2351ed..a6bb98a 100644 --- a/src/dhcpsrv/app.py +++ b/src/dhcpsrv/app.py @@ -21,7 +21,7 @@ from .config import load_config from .i18n import set_language, t -def _select_nic(console: Console) -> dict | None: +def _select_nic(console: Console, auto_single: bool = True) -> dict | None: console.rule(f"[bold cyan]{t('available_adapters')}") adapters = list_adapters() if not adapters: @@ -30,6 +30,10 @@ def _select_nic(console: Console) -> dict | None: for i, a in enumerate(adapters, 1): ip = a.get("IPv4") or "—" console.print(f" {i}) [{a['Status']}] {a['Name']} ({a['Description']}) {ip}") + if auto_single and len(adapters) == 1: + only = adapters[0] + console.print(f"[green]{t('only_adapter', name=only['Name'])}[/]") + return only while True: s = Prompt.ask(t("select_adapter")).strip() if s.isdigit() and 1 <= int(s) <= len(adapters): @@ -64,7 +68,7 @@ def main() -> None: console.print(title) console.print() - nic = _select_nic(console) + nic = _select_nic(console, cfg_data.get("auto_select_single", True)) if not nic: input(t("press_enter")); return diff --git a/src/dhcpsrv/config.py b/src/dhcpsrv/config.py index 1b1c9e4..7dc9252 100644 --- a/src/dhcpsrv/config.py +++ b/src/dhcpsrv/config.py @@ -20,16 +20,25 @@ CONFIG_HEADER = """\ # --------------------------------------------------------------------------- # dhcpsrv configuration # -# To change the interface language, edit the 'language' value below. -# Valid values: en, ru +# language : interface language. Valid: en, ru +# auto_select_single : auto-pick the network adapter when only one is found, +# without asking (yes/no). # -# Чтобы сменить язык интерфейса, измените значение 'language' ниже. -# Допустимые значения: en, ru +# language : язык интерфейса. Допустимо: en, ru +# auto_select_single : автоматически выбирать сетевой адаптер, если он +# единственный, не спрашивая (yes/no). # --------------------------------------------------------------------------- """ +def _to_bool(s: str, default: bool) -> bool: + s = (s or "").strip().lower() + if s in ("yes", "true", "y", "1", "on"): return True + if s in ("no", "false", "n", "0", "off"): return False + return default + + def app_dir() -> Path: """Directory holding the running executable (or source folder when run via python).""" if getattr(sys, "frozen", False): @@ -60,10 +69,26 @@ def _ask_language() -> str: print("Please enter 1 or 2 / Введите 1 или 2") -def _write_config(lang: str) -> None: +def _ask_auto_select() -> bool: + """First-run prompt — auto-select a lone adapter? Bilingual stdin.""" + print() + print("Auto-select the network adapter when only one is found? (y/n)") + print("Автовыбор сетевого адаптера, если он единственный? (y/n)") + while True: + try: + c = input("> ").strip().lower() + except (EOFError, KeyboardInterrupt): + return True + if c in ("y", "yes", "д", "да", "1"): return True + if c in ("n", "no", "н", "нет", "0"): return False + print("y / n") + + +def _write_config(lang: str, auto_select: bool) -> None: path = config_path() cp = configparser.ConfigParser() - cp["General"] = {"language": lang} + cp["General"] = {"language": lang, + "auto_select_single": "yes" if auto_select else "no"} with path.open("w", encoding="utf-8") as f: f.write(CONFIG_HEADER) cp.write(f) @@ -75,20 +100,22 @@ def load_config() -> dict: path = config_path() if not path.exists(): lang = _ask_language() + auto = _ask_auto_select() try: - _write_config(lang) + _write_config(lang, auto) except OSError: # read-only location — fall back to in-memory default pass - return {"language": lang} + return {"language": lang, "auto_select_single": auto} cp = configparser.ConfigParser() try: cp.read(path, encoding="utf-8") except (configparser.Error, OSError): - return {"language": DEFAULT_LANG} + return {"language": DEFAULT_LANG, "auto_select_single": True} lang = (cp.get("General", "language", fallback=DEFAULT_LANG) or DEFAULT_LANG).strip().lower() if lang not in SUPPORTED_LANGS: lang = DEFAULT_LANG - return {"language": lang} + auto = _to_bool(cp.get("General", "auto_select_single", fallback="yes"), True) + return {"language": lang, "auto_select_single": auto} diff --git a/src/dhcpsrv/dhcp.py b/src/dhcpsrv/dhcp.py index cd09ba5..8a17e1e 100644 --- a/src/dhcpsrv/dhcp.py +++ b/src/dhcpsrv/dhcp.py @@ -16,6 +16,7 @@ from typing import Callable, Optional from .network import ping_one from .i18n import t as _t +from .leases import write_leases # ---------- helpers ---------- @@ -147,6 +148,23 @@ class DhcpServer: o += bytes([255]) return bytes(pkt) + bytes(o) + # --- lease-table export (contract for sibling tools, e.g. vrcx) --- + def export_leases(self) -> None: + """Snapshot the lease table to the shared JSON file. Best-effort.""" + with self.lock: + leases = [ + { + "ip": int2ip(c["ip_int"]), + "mac": mac, + "host": c.get("host", ""), + "ping_ok": bool(c.get("ping_ok")), + "last": c.get("last", ""), + } + for mac, c in self.clients.items() + if c.get("ip_int") + ] + write_leases(self.cfg.server_ip, leases, now_s()) + # --- ping loop, runs in its own thread --- def ping_loop(self, stop: threading.Event) -> None: from concurrent.futures import ThreadPoolExecutor @@ -166,6 +184,8 @@ class DhcpServer: changed = True if changed: self.on_change() + # Refresh the export every tick so its mtime reflects liveness. + self.export_leases() # --- main server loop --- def run(self, stop: threading.Event) -> None: diff --git a/src/dhcpsrv/i18n.py b/src/dhcpsrv/i18n.py index abe561d..da78dcc 100644 --- a/src/dhcpsrv/i18n.py +++ b/src/dhcpsrv/i18n.py @@ -23,6 +23,7 @@ STRINGS: dict[str, dict[str, str]] = { "update_available": "Update available ({tag})", "available_adapters": "Available adapters", "no_adapters": "No suitable wired adapters found.", + "only_adapter": "Only one adapter, auto-selected: {name}", "select_adapter": "Select adapter number", "invalid_selection": "Invalid selection.", "press_enter": "Press Enter to exit", @@ -60,6 +61,7 @@ STRINGS: dict[str, dict[str, str]] = { "update_available": "Доступно обновление ({tag})", "available_adapters": "Доступные адаптеры", "no_adapters": "Подходящие проводные адаптеры не найдены.", + "only_adapter": "Адаптер один, выбран автоматически: {name}", "select_adapter": "Введите номер адаптера", "invalid_selection": "Неверный выбор.", "press_enter": "Нажмите Enter для выхода", diff --git a/src/dhcpsrv/leases.py b/src/dhcpsrv/leases.py new file mode 100644 index 0000000..b420383 --- /dev/null +++ b/src/dhcpsrv/leases.py @@ -0,0 +1,51 @@ +""" +Lease-table export — the shared contract with sibling tools (e.g. vrcx). + +dhcpsrv owns the DHCP state; other tools must NOT reach into it. Instead we +periodically dump a small, self-describing JSON snapshot to a fixed, +tool-neutral location. Consumers read that file (and judge its freshness by +the file mtime). No shared code, no network call between the tools. + +Default location: %PROGRAMDATA%\\vegman\\leases.json (override via the +VEGMAN_LEASES env var). The directory is created on first write. +""" + +from __future__ import annotations +import json +import os +import tempfile +from pathlib import Path + +SCHEMA = 1 + + +def leases_path() -> Path: + override = os.environ.get("VEGMAN_LEASES") + if override: + return Path(override) + base = os.environ.get("PROGRAMDATA") or os.environ.get("LOCALAPPDATA") or "." + return Path(base) / "vegman" / "leases.json" + + +def write_leases(server_ip: str, leases: list[dict], updated: str) -> None: + """Atomically write the snapshot. Best-effort: swallow I/O errors so a + read-only/locked target can never disturb the DHCP service.""" + path = leases_path() + payload = { + "schema": SCHEMA, + "server_ip": server_ip, + "updated": updated, + "leases": leases, + } + try: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".leases-", suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False) + os.replace(tmp, path) + finally: + if os.path.exists(tmp): + os.remove(tmp) + except OSError: + pass