253 lines
7.8 KiB
Python
253 lines
7.8 KiB
Python
"""Device registration and token persistence for pypretix-device."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import platform
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
import requests
|
|
|
|
from pypretix_device.models import DeviceInfo
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CONFIG_FILENAME = "config.json"
|
|
SOFTWARE_BRAND = "pypretix"
|
|
SOFTWARE_VERSION = "0.1.0"
|
|
|
|
|
|
def _get_config_path() -> Path:
|
|
"""Get the config file path."""
|
|
from pypretix_device.models import CONFIG_DIR
|
|
if CONFIG_DIR:
|
|
return Path(CONFIG_DIR) / CONFIG_FILENAME
|
|
return Path.home() / ".config" / "pypretix" / CONFIG_FILENAME
|
|
|
|
|
|
def _ensure_config_dir() -> Path:
|
|
"""Create config directory if it doesn't exist."""
|
|
config_path = _get_config_path().parent
|
|
config_path.mkdir(parents=True, exist_ok=True)
|
|
return config_path
|
|
|
|
|
|
def load_config() -> Optional[Dict[str, Any]]:
|
|
"""Load config from disk. Returns None if not found or invalid."""
|
|
config_path = _get_config_path()
|
|
if not config_path.exists():
|
|
return None
|
|
try:
|
|
with open(config_path, "r") as f:
|
|
data = json.load(f)
|
|
# Validate required fields
|
|
required = ["base_url", "api_token"]
|
|
if not all(k in data for k in required):
|
|
logger.warning("Config file missing required fields")
|
|
return None
|
|
return data
|
|
except (json.JSONDecodeError, OSError) as e:
|
|
logger.error(f"Failed to load config: {e}")
|
|
return None
|
|
|
|
|
|
def save_config(config: Dict[str, Any]) -> None:
|
|
"""Save config to disk."""
|
|
_ensure_config_dir()
|
|
config_path = _get_config_path()
|
|
with open(config_path, "w") as f:
|
|
json.dump(config, f, indent=2)
|
|
logger.info(f"Config saved to {config_path}")
|
|
|
|
|
|
class RegistrationFailedError(Exception):
|
|
"""Raised when device registration fails."""
|
|
pass
|
|
|
|
|
|
def register_device(
|
|
base_url: str,
|
|
init_token: str,
|
|
device_name: str = "example-scanner",
|
|
) -> DeviceInfo:
|
|
"""Register a new device with pretix using an initialization token.
|
|
|
|
Args:
|
|
base_url: Base URL of the pretix instance (e.g. "http://localhost").
|
|
init_token: The 24-character initialization token from the web UI.
|
|
device_name: Human-readable name for this device.
|
|
|
|
Returns:
|
|
DeviceInfo with the newly assigned API token.
|
|
|
|
Raises:
|
|
RegistrationFailedError: If registration fails.
|
|
"""
|
|
# Normalize base_url - ensure no trailing slash, ensure no path
|
|
base_url = base_url.rstrip("/")
|
|
url = f"{base_url}/api/v1/device/initialize"
|
|
|
|
payload = {
|
|
"token": init_token,
|
|
"hardware_brand": platform.machine() or "unknown",
|
|
"hardware_model": "pypretix-device",
|
|
"os_name": platform.system() or "unknown",
|
|
"os_version": platform.version() or "unknown",
|
|
"software_brand": SOFTWARE_BRAND,
|
|
"software_version": SOFTWARE_VERSION,
|
|
}
|
|
|
|
try:
|
|
resp = requests.post(url, json=payload, timeout=30)
|
|
except requests.RequestException as e:
|
|
raise RegistrationFailedError(f"Failed to connect to pretix at {base_url}: {e}")
|
|
|
|
if resp.status_code != 200:
|
|
try:
|
|
errors = resp.json()
|
|
error_msg = json.dumps(errors)
|
|
except (ValueError, KeyError):
|
|
error_msg = resp.text[:500]
|
|
raise RegistrationFailedError(
|
|
f"Device registration failed (HTTP {resp.status_code}): {error_msg}"
|
|
)
|
|
|
|
data = resp.json()
|
|
device_info = DeviceInfo(
|
|
organizer=data["organizer"],
|
|
device_id=data["device_id"],
|
|
name=data["name"],
|
|
api_token=data["api_token"],
|
|
unique_serial=data.get("unique_serial", ""),
|
|
gate=data.get("gate"),
|
|
server_version=data.get("server", {}).get("version", {}).get("pretix"),
|
|
)
|
|
|
|
# Save config if directory is set
|
|
from pypretix_device.models import CONFIG_DIR
|
|
if CONFIG_DIR:
|
|
config = {
|
|
"base_url": base_url,
|
|
"api_token": device_info.api_token,
|
|
"device_id": device_info.device_id,
|
|
"organizer": device_info.organizer,
|
|
"device_name": device_name,
|
|
"software_version": SOFTWARE_VERSION,
|
|
}
|
|
if data.get("gate"):
|
|
config["checkin_list"] = data["gate"].get("id")
|
|
save_config(config)
|
|
logger.info(f"Device '{device_info.name}' registered successfully (id={device_info.device_id})")
|
|
|
|
return device_info
|
|
|
|
|
|
def get_or_register_device(
|
|
base_url: str,
|
|
init_token: Optional[str] = None,
|
|
device_name: str = "helferscanner",
|
|
) -> DeviceInfo:
|
|
"""Load existing device config or register a new one.
|
|
|
|
Args:
|
|
base_url: Base URL of the pretix instance.
|
|
init_token: Init token for registration. Required if no config exists.
|
|
device_name: Device name for registration.
|
|
|
|
Returns:
|
|
DeviceInfo for the current device.
|
|
|
|
Raises:
|
|
RegistrationFailedError: If no config exists and no init_token provided.
|
|
"""
|
|
config = load_config()
|
|
if config and "api_token" in config:
|
|
logger.info(f"Loaded config from {_get_config_path()}")
|
|
return DeviceInfo(
|
|
organizer=config.get("organizer", ""),
|
|
device_id=config.get("device_id", 0),
|
|
name=config.get("device_name", "Unknown"),
|
|
api_token=config["api_token"],
|
|
unique_serial=config.get("unique_serial", ""),
|
|
)
|
|
|
|
if not init_token:
|
|
raise RegistrationFailedError(
|
|
"No existing config found and no init_token provided.\n"
|
|
"Create a new device in the pretix web UI (Organizer Settings => Devices) "
|
|
"to obtain an initialization token."
|
|
)
|
|
|
|
return register_device(base_url, init_token, device_name)
|
|
|
|
|
|
def revoke_device() -> None:
|
|
"""Revoke (disable) the current device. This invalidates the API token permanently."""
|
|
config = load_config()
|
|
if not config:
|
|
logger.warning("No config found, nothing to revoke")
|
|
return
|
|
|
|
base_url = config.get("base_url", "").rstrip("/")
|
|
api_token = config.get("api_token", "")
|
|
|
|
if not base_url or not api_token:
|
|
logger.warning("Config missing base_url or api_token")
|
|
return
|
|
|
|
url = f"{base_url}/api/v1/device/revoke/"
|
|
try:
|
|
resp = requests.post(
|
|
url,
|
|
headers={"Authorization": f"Device {api_token}"},
|
|
timeout=10,
|
|
)
|
|
if resp.status_code in (200, 204):
|
|
logger.info("Device revoked successfully")
|
|
else:
|
|
logger.error(f"Revocation failed (HTTP {resp.status_code})")
|
|
except requests.RequestException as e:
|
|
logger.error(f"Failed to revoke device: {e}")
|
|
|
|
# Remove config file
|
|
config_path = _get_config_path()
|
|
if config_path.exists():
|
|
config_path.unlink()
|
|
logger.info(f"Config file removed: {config_path}")
|
|
|
|
|
|
def update_device_info(
|
|
base_url: str,
|
|
api_token: str,
|
|
software_version: str = SOFTWARE_VERSION,
|
|
) -> Dict[str, Any]:
|
|
"""Notify the server about a software version update."""
|
|
base_url = base_url.rstrip("/")
|
|
url = f"{base_url}/api/v1/device/update/"
|
|
|
|
payload = {
|
|
"hardware_brand": platform.machine() or "unknown",
|
|
"hardware_model": "pypretix-device",
|
|
"os_name": platform.system() or "unknown",
|
|
"os_version": platform.version() or "unknown",
|
|
"software_brand": SOFTWARE_BRAND,
|
|
"software_version": software_version,
|
|
}
|
|
|
|
try:
|
|
resp = requests.post(
|
|
url,
|
|
headers={"Authorization": f"Device {api_token}"},
|
|
json=payload,
|
|
timeout=10,
|
|
)
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
logger.debug(f"Update failed (HTTP {resp.status_code})")
|
|
return {}
|
|
except requests.RequestException as e:
|
|
logger.debug(f"Update request failed: {e}")
|
|
return {}
|