MAINHACK

Viewing / Editing File: remote_iplist.py

File is not writable. Editing disabled.
import asyncio
import logging
import urllib
import json
import gzip
from http.client import RemoteDisconnected
from typing import Iterable

from defence360agent.api.server import API, APIError

logger = logging.getLogger(__name__)


class RemoteIPListAPI(API):
    URL = "/api/sync/v1/iplist"

    @classmethod
    async def sync(
        cls, server_id, purposes, sync_point, checksum, cln_license
    ) -> Iterable[dict]:
        loop = asyncio.get_event_loop()
        try:
            request = urllib.request.Request(
                cls._BASE_URL + cls.URL,
                method="POST",
                data=json.dumps(
                    {
                        "sync_point": sync_point,
                        "server_id": server_id,
                        "purposes": purposes,
                        "checksum": checksum,
                        "license": cln_license,
                    }
                ).encode(),
                headers={"Content-Type": "application/json"},
            )
            raw_response = await loop.run_in_executor(
                None,
                cls.request,
                request,
                False,
            )
            # urllib does not support chunked read of chunked response
            return cls._lazy_chunks_iterator(raw_response)
        except (
            asyncio.TimeoutError,
            RemoteDisconnected,
            ValueError,
            APIError,
        ) as e:
            logger.error("Failed to get iplist update from server: %s", e)
            return []

    @classmethod
    def _lazy_chunks_iterator(cls, raw_response):
        for chunk in gzip.decompress(raw_response).split(b"\n"):
            if chunk:
                yield json.loads(chunk)
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: remote_iplist.py

Server Info

System Features

User Info
