send_validation_request.py

"""Send a validation request to the validator endpoint.
"""

import asyncio
import os
import ssl
import sys
from typing import Final

import certifi
import websockets

from generic_args import parse_args

VALIDATOR_URI: Final[str] = (
    "wss://127.0.0.1:8001/ws/validate/ee4eed14-ffc2-11ed-9f67-67fb68ae3988/"
)


async def secure_local() -> ssl.SSLContext:
    """Use a local certificate/public-key to connect to a validator
    server.
    """
    pem_file = os.environ.get("LOCAL_CERT", "validator.pem")
    if not os.path.exists(pem_file):
        print("ensure that the pem file exists:", pem_file, file=sys.stderr)
        sys.exit(1)
    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ssl_context.load_verify_locations(pem_file)
    await send_validation_request(local=False, secure=False, local_secure=ssl_context)


async def send_validation_request(
    local: bool = False,
    secure: bool = False,
    local_secure: ssl.SSLContext = None,
):
    """Send a validation request to the validation API and print the
    result.
    """

    validator_connection = f"{VALIDATOR_URI}"

    cert = None
    if secure:
        cert = ssl.create_default_context(cafile=certifi.where())
    elif local_secure:
        cert = local_secure
    elif local:
        cert = None
    else:
        print(
            "no connection method specified. use `--help` for more information",
            file=sys.stderr,
        )
        sys.exit(1)

    # pylint: disable=E1101
    async with websockets.connect(validator_connection, ssl=cert) as websocket:
        await websocket.send("PING")
        msg = await websocket.recv()
        print(msg)


async def main():
    """Primary entry point of this script."""

    args = parse_args(
        "fetch and store",
        "fetch and store data from collector node REST APIs (default: remote over ssl)",
    )
    if args.local:
        await send_validation_request(local=True)
    elif args.secure_local:
        # use a local certificate to connect to the validator (local).
        await secure_local()
    else:
        # use a certificate authority to connect to the validator (remote).
        await send_validation_request(secure=True)


if __name__ == "__main__":
    asyncio.run(main())