| Server IP : 172.67.174.47 / Your IP : 216.73.216.87 Web Server : LiteSpeed System : Linux premium151.web-hosting.com 4.18.0-553.44.1.lve.el8.x86_64 #1 SMP Thu Mar 13 14:29:12 UTC 2025 x86_64 User : tempvsty ( 647) PHP Version : 8.0.30 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /././opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/api/server/ |
Upload File : |
import asyncio
import http.client
import json
import logging
import socket
import urllib.error
import urllib.request
from defence360agent.contracts.config import Core
logger = logging.getLogger(__name__)
class APIError(Exception):
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
if len(args) >= 2:
_, status_code, *args = args
self.status_code = status_code
else:
self.status_code = None
class APIErrorTooManyRequests(APIError):
...
class APITokenError(APIError):
...
class FGWSendMessgeException(Exception):
...
class API:
_BASE_URL = Core.API_BASE_URL
# socket timeout is for blocking operations
# it should be as less as possible, but for sync api (remote_iplist)
# we may wait for response for 25 seconds, let's set it to 45 from our side
_SOCKET_TIMEOUT = 45
@classmethod
def request(cls, request: urllib.request.Request, json_loads=True):
try:
with urllib.request.urlopen(
request,
# agent should be able to wait for a while
# in lb queue before being connected
timeout=cls._SOCKET_TIMEOUT,
) as response:
logger.info(
"Performed request for url=%s method=%s body size=%s"
" status=%s",
request.full_url,
getattr(request, "method", None),
len(request.data) if request.data else 0,
response.status,
)
if response.status != 200:
raise APIError(
"status code is {}".format(response.status),
response.status,
)
plain_response = response.read()
logger.info("Response=%s ...", plain_response[:50])
if json_loads:
result = json.loads(plain_response.decode())
else:
result = plain_response
return result
except (
UnicodeDecodeError,
http.client.HTTPException,
json.JSONDecodeError,
socket.timeout,
urllib.error.URLError,
) as e:
status_code = getattr(e, "code", None)
if status_code == 429:
raise APIErrorTooManyRequests(
"request failed, reason: %s" % (e,), status_code
) from e
raise APIError(
"request failed, reason: %s" % (e,), status_code
) from e
@classmethod
async def async_request(cls, request, executor=None):
loop = asyncio.get_event_loop()
return await loop.run_in_executor(executor, cls.request, request)