miracast-client/linux/wfd/wpa.py

"""Steuerung von wpa_supplicant über seinen Control-Socket.

Das ist derselbe Kanal, den auch wpa_cli benutzt: ein UNIX-Datagramm-Socket,
über den Befehle im Klartext gehen. Ereignisse (Gerät gefunden, Gruppe
gestartet) kommen über eine zweite, angemeldete Verbindung herein.
"""

import logging
import os
import queue
import socket
import tempfile
import threading
from typing import Dict, List, Optional

log = logging.getLogger("wfd.wpa")

CONTROL_DIRS = ("/run/wpa_supplicant", "/var/run/wpa_supplicant")


def find_interfaces() -> List[str]:
    """Alle Interfaces, für die ein Control-Socket bereitsteht."""
    found = []
    for directory in CONTROL_DIRS:
        try:
            for name in sorted(os.listdir(directory)):
                path = os.path.join(directory, name)
                if os.path.exists(path) and path not in found:
                    found.append(path)
        except OSError:
            continue
    return found


def wireless_interfaces() -> List[str]:
    """Namen der WLAN-Geräte laut Kernel."""
    try:
        return sorted(os.listdir("/sys/class/net")) and [
            name for name in sorted(os.listdir("/sys/class/net"))
            if os.path.exists(f"/sys/class/net/{name}/wireless")
            or os.path.exists(f"/sys/class/net/{name}/phy80211")
        ]
    except OSError:
        return []


class WpaError(RuntimeError):
    pass


class WpaClient:
    """Eine Verbindung zum Control-Socket eines Interfaces."""

    def __init__(self, control_path: str):
        self.control_path = control_path
        self._local_path = os.path.join(
            tempfile.gettempdir(), f"wfd-ctrl-{os.getpid()}-{id(self)}"
        )
        self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        try:
            self._sock.bind(self._local_path)
            self._sock.connect(control_path)
        except OSError as error:
            self._cleanup()
            raise WpaError(
                f"Kein Zugriff auf {control_path}: {error}. "
                "Meist fehlen Rechte - als root ausführen oder der Gruppe netdev beitreten."
            ) from error
        self._sock.settimeout(5)
        self._lock = threading.Lock()

        self.events: "queue.Queue[str]" = queue.Queue()
        self._attached = False
        self._reader: Optional[threading.Thread] = None
        self._stop = threading.Event()

    # ------------------------------------------------------------ Befehle

    def request(self, command: str, timeout: float = 5.0) -> str:
        """Schickt einen Befehl und liefert die Antwort."""
        with self._lock:
            self._sock.settimeout(timeout)
            self._sock.send(command.encode())
            while True:
                data = self._sock.recv(8192).decode("utf-8", "replace")
                # Ereignisse können dazwischenfunken - die gehören in die Queue.
                if data.startswith("<"):
                    self.events.put(data)
                    continue
                return data.strip()

    def ok(self, command: str) -> bool:
        try:
            return self.request(command).startswith("OK")
        except OSError as error:
            log.warning("Befehl '%s' fehlgeschlagen: %s", command, error)
            return False

    # ---------------------------------------------------------- Ereignisse

    def attach(self):
        """Meldet sich für Ereignisse an und liest sie im Hintergrund mit."""
        if self._attached:
            return
        if not self.request("ATTACH").startswith("OK"):
            raise WpaError("wpa_supplicant nimmt keine Ereignis-Anmeldung an")
        self._attached = True
        self._reader = threading.Thread(target=self._read_events, daemon=True,
                                        name="wpa-events")
        self._reader.start()

    def _read_events(self):
        while not self._stop.is_set():
            try:
                with self._lock:
                    self._sock.settimeout(0.4)
                    data = self._sock.recv(8192).decode("utf-8", "replace")
            except socket.timeout:
                continue
            except OSError:
                break
            for line in data.splitlines():
                if line.startswith("<"):
                    # Prioritätsangabe wie "<3>" abtrennen
                    self.events.put(line.split(">", 1)[-1])
                    log.debug("Ereignis: %s", line)

    def wait_for_event(self, *needles: str, timeout: float = 30.0) -> Optional[str]:
        """Wartet auf ein Ereignis, das eine der Zeichenketten enthält."""
        import time
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            try:
                event = self.events.get(timeout=0.3)
            except queue.Empty:
                continue
            if any(needle in event for needle in needles):
                return event
        return None

    def close(self):
        self._stop.set()
        if self._reader:
            self._reader.join(1.0)
        try:
            self._sock.close()
        except OSError:
            pass
        self._cleanup()

    def _cleanup(self):
        try:
            os.unlink(self._local_path)
        except OSError:
            pass


def parse_key_values(text: str) -> Dict[str, str]:
    """Wandelt die übliche 'schlüssel=wert'-Ausgabe in ein Wörterbuch."""
    result = {}
    for line in text.splitlines():
        if "=" in line:
            key, value = line.split("=", 1)
            result[key.strip()] = value.strip()
    return result