yaris-obd/modules/app_remote_config.py

"""Manage remote server credentials stored in db/upload_creds.env."""

import json
import os
from urllib.parse import urlparse

import requests
from dotenv import dotenv_values


_BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_CREDS_PATH = os.path.join(_BASE_DIR, "db", "upload_creds.env")
_ORDERED_KEYS = [
    "YARIS_SERVER_URL",
    "YARIS_DEVICE_NAME",
    "YARIS_API_KEY",
    "YARIS_CELLULAR_ENABLED",
    "YARIS_CELLULAR_APN",
    "YARIS_CELLULAR_APN_USER",
    "YARIS_CELLULAR_APN_PASS",
]


def server_credentials():
    values = _read_values()
    return {
        "server_url": values.get("YARIS_SERVER_URL", "").rstrip("/"),
        "device_name": values.get("YARIS_DEVICE_NAME", ""),
        "api_key": values.get("YARIS_API_KEY", ""),
    }


def get_remote_config():
    credentials = server_credentials()
    return {
        "server_url": credentials["server_url"],
        "device_name": credentials["device_name"],
        "api_key_configured": bool(credentials["api_key"]),
    }


def update_remote_config(body, errors):
    if not isinstance(body, dict):
        errors["error"] = "Invalid request body"
        return errors

    values = _read_values()

    if "server_url" in body:
        server_url = str(body["server_url"]).strip()
        if server_url:
            parsed = urlparse(server_url)
            if parsed.scheme not in {"http", "https"} or not parsed.netloc:
                errors["server_url"] = "Server URL must be a valid http(s) URL."
            else:
                values["YARIS_SERVER_URL"] = server_url.rstrip("/")
        else:
            values["YARIS_SERVER_URL"] = ""

    if "device_name" in body:
        device_name = str(body["device_name"]).strip()
        if device_name and len(device_name) > 100:
            errors["remote_device_name"] = "Device name must be 1-100 characters."
        else:
            values["YARIS_DEVICE_NAME"] = device_name

    if body.get("clear_api_key"):
        values["YARIS_API_KEY"] = ""
    elif "api_key" in body:
        api_key = str(body["api_key"]).strip()
        if api_key:
            if len(api_key) > 512:
                errors["api_key"] = "API key must be 1-512 characters."
            else:
                values["YARIS_API_KEY"] = api_key

    if errors:
        return errors

    _write_values(values)
    return errors


def test_connection():
    credentials = server_credentials()
    if not credentials["server_url"]:
        return {"ok": False, "error": "Server URL is not configured."}
    try:
        health_response = requests.get(
            f"{credentials['server_url']}/health",
            timeout=5,
        )
        health_response.raise_for_status()
    except requests.RequestException as error:
        return {"ok": False, "error": f"Could not reach server: {error}"}

    if not credentials["device_name"] or not credentials["api_key"]:
        return {
            "ok": True,
            "message": "Server is reachable, but device credentials are not fully configured yet.",
            "authenticated": False,
        }

    try:
        auth_response = requests.get(
            f"{credentials['server_url']}/api/device-config",
            headers=_headers(credentials),
            timeout=5,
        )
    except requests.RequestException as error:
        return {"ok": False, "error": f"Server reachable but auth test failed: {error}"}

    if auth_response.status_code == 200:
        return {
            "ok": True,
            "message": "Server reachable and device credentials are valid.",
            "authenticated": True,
        }
    if auth_response.status_code == 401:
        return {
            "ok": False,
            "error": "Server reachable, but device name or API key is invalid.",
        }
    return {
        "ok": False,
        "error": f"Server reachable, but config auth returned HTTP {auth_response.status_code}.",
    }


def _headers(credentials):
    return {
        "X-Device-Name": credentials["device_name"],
        "X-Api-Key": credentials["api_key"],
    }


def _read_values():
    values = {key: "" for key in _ORDERED_KEYS}
    parsed = dotenv_values(_CREDS_PATH)
    for key, value in parsed.items():
        values[key] = value or ""
    return values


def _write_values(values):
    os.makedirs(os.path.dirname(_CREDS_PATH), exist_ok=True)
    with open(_CREDS_PATH, "w", encoding="utf-8") as handle:
        handle.write("##\n")
        handle.write("## yaris-obd remote server upload client credentials\n")
        handle.write("##\n")
        handle.write("## Managed via the local Settings page. Real values stay local to this Pi.\n")
        handle.write("##\n\n")
        for key in _ORDERED_KEYS:
            handle.write(f"{key} = {json.dumps(values.get(key, ''))}\n")