checktheroads-hyperglass/hyperglass/api/tasks.py

"""Tasks to be executed from web API."""

# Standard Library
from typing import Dict, Union
from pathlib import Path

# Third Party
from httpx import Headers


def import_public_key(
    app_path: Union[Path, str], device_name: str, keystring: str
) -> bool:
    """Import a public key for hyperglass-agent."""
    if not isinstance(app_path, Path):
        app_path = Path(app_path)

    cert_dir = app_path / "certs"

    if not cert_dir.exists():
        cert_dir.mkdir()

    if not cert_dir.exists():
        raise RuntimeError(f"Failed to create certs directory at {str(cert_dir)}")

    filename = f"{device_name}.pem"
    cert_file = cert_dir / filename

    with cert_file.open("w+") as file:
        file.write(str(keystring))

    with cert_file.open("r") as file:
        read_file = file.read().strip()
        if not keystring == read_file:
            raise RuntimeError("Wrote key, but written file did not match input key")

    return True


async def process_headers(headers: Headers) -> Dict:
    """Filter out unwanted headers and return as a dictionary."""
    headers = dict(headers)
    header_keys = (
        "user-agent",
        "referer",
        "accept-encoding",
        "accept-language",
        "x-real-ip",
        "x-forwarded-for",
    )
    return {k: headers.get(k) for k in header_keys}