Upload 5 files

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.11-slim
+
+RUN useradd -m -u 1000 user && \
+    mkdir /home/user/app && \
+    chown -R user:user /home/user/app
+
+WORKDIR /home/user/app
+
+RUN apt-get update && \
+    apt-get install -y && \
+    rm -rf /var/lib/apt/lists/*
+
+COPY --chown=user:user ./requirements.txt ./
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+USER user
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH
+
+COPY --chown=user:user . .
+
+CMD ["python", "main.py"]

connection_manager.py

@@ -0,0 +1,24 @@
+from typing import List
+
+from fastapi import WebSocket, WebSocketException
+
+
+class ConnectionManager:
+    def __init__(self, password: str) -> None:
+        self.__password: str = password
+        self.__active_connections: List[WebSocket] = list[WebSocket]()
+
+    async def connect(self, websocket: WebSocket) -> None:
+        authorization: str = websocket.headers.get("Authorization")
+        if not authorization:
+            raise WebSocketException(code=1008, reason="Authorization header missing")
+        token_type, _, token = authorization.partition(' ')
+        if token_type != "Bearer" or not token:
+            raise WebSocketException(code=1008, reason="Invalid authorization header format")
+        if token != self.__password:
+            raise WebSocketException(code=1008, reason="Invalid token")
+        await websocket.accept()
+        self.__active_connections.append(websocket)
+
+    def disconnect(self, websocket: WebSocket) -> None:
+        self.__active_connections.remove(websocket)

main.py

@@ -0,0 +1,78 @@
+import os
+import uvicorn
+from fastapi import FastAPI, Request, HTTPException, status, WebSocketDisconnect
+from fastapi.applications import AppType
+from fastapi.responses import JSONResponse
+from requests import Timeout, TooManyRedirects, RequestException, Session
+
+from connection_manager import *
+from request_forwarder_classes import *
+
+PORT: int = int(os.getenv("PORT"))
+PASSWORD: str = os.getenv("PASSWORD")
+
+app: AppType = FastAPI()
+connection_manager = ConnectionManager(PASSWORD)
+
+
+def __get_response(forwarded_request: ForwardedRequest, session: Session | None = None) -> Tuple[int, Dict]:
+    try:
+        if not session:
+            session = Session()
+        response = session.request(
+            forwarded_request.method,
+            forwarded_request.url,
+            headers=forwarded_request.headers,
+            data=forwarded_request.data,
+            params=forwarded_request.params,
+            auth=forwarded_request.auth,
+            cookies=forwarded_request.cookies,
+            json=forwarded_request.json)
+        forwarded_response = ForwardedResponse.from_response(response, forwarded_request.response_content_type)
+        return status.HTTP_200_OK, {"response": forwarded_response.to_json_dict(), "error": None}
+    except ConnectionError as e:
+        return status.HTTP_418_IM_A_TEAPOT, {"response": None, "error": {"ConnectionError": str(e)}}
+    except Timeout as e:
+        return status.HTTP_418_IM_A_TEAPOT, {"response": None, "error": {"Timeout": str(e)}}
+    except TooManyRedirects as e:
+        return status.HTTP_418_IM_A_TEAPOT, {"response": None, "error": {"TooManyRedirects": str(e)}}
+    except RequestException as e:
+        return status.HTTP_418_IM_A_TEAPOT, {"response": None, "error": {"RequestException": str(e)}}
+    except Exception as e:
+        return status.HTTP_418_IM_A_TEAPOT, {"response": None, "error": str(e)}
+
+
+@app.websocket("/ws/forward-requests")
+async def websocket_endpoint(websocket: WebSocket):
+    await connection_manager.connect(websocket)
+    session = Session()
+    try:
+        while True:
+            request = await websocket.receive_json()
+            print(request)
+            forwarded_request = ForwardedRequest.from_json(request)
+            response_status, response = __get_response(forwarded_request, session)
+            await websocket.send_json(response)
+    except WebSocketDisconnect:
+        connection_manager.disconnect(websocket)
+
+
+@app.post("/forward-request")
+async def forward_request(forwarded_request: ForwardedRequest, request: Request):
+    authorization: str = request.headers.get("Authorization")
+    if not authorization:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Authorization header missing")
+
+    token_type, _, token = authorization.partition(' ')
+    if token_type != "Bearer" or not token:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authorization header format")
+
+    if token != PASSWORD:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid token")
+
+    response_status, response = __get_response(forwarded_request)
+    return JSONResponse(content=response, status_code=response_status)
+
+
+if __name__ == '__main__':
+    uvicorn.run("main:app", host="0.0.0.0", port=PORT)

request_forwarder_classes.py

@@ -0,0 +1,129 @@
+import datetime
+from typing import Any, Dict, Tuple, Literal
+
+import requests
+from pydantic import BaseModel
+
+RequestMethod = Literal['GET', 'POST', 'PUT', 'DELETE', 'HEAD', 'OPTIONS', 'PATCH']
+ResponseType = Literal['raw', 'text', 'json']
+
+
+class ForwardedRequest(BaseModel):
+    method: RequestMethod
+    url: str
+    headers: Dict[str, str] | None = None
+    data: Dict[str, Any] | Tuple[Tuple[str, Any]] | str | bytes | None = None
+    params: Dict[str, Any] | Tuple[Tuple[str, Any]] | str | None = None
+    auth: Tuple[str, str] | None = None
+    cookies: Dict[str, str] | None = None
+    json: Any | None = None
+    response_content_type: ResponseType = 'text'
+
+    @staticmethod
+    def from_request(request: requests.Request, response_content_type: ResponseType = 'text') -> 'ForwardedRequest':
+        return ForwardedRequest(
+            method=request.method,
+            url=request.url,
+            headers=request.headers,
+            data=request.data,
+            params=request.params,
+            auth=request.auth,
+            cookies=request.cookies,
+            json=request.json,
+            response_content_type=response_content_type
+        )
+
+    @staticmethod
+    def from_json(json_dict: Dict[str, Any]) -> 'ForwardedRequest':
+        return ForwardedRequest(
+            method=json_dict['method'],
+            url=json_dict['url'],
+            headers=json_dict['headers'],
+            data=json_dict['data'],
+            params=json_dict['params'],
+            auth=json_dict['auth'],
+            cookies=json_dict['cookies'],
+            json=json_dict['json'],
+            response_content_type=json_dict['response_content_type'],
+        )
+
+    def to_json_dict(self):
+        return {
+            "method": self.method,
+            "url": self.url,
+            "headers": self.headers,
+            "data": self.data,
+            "params": self.params,
+            "auth": self.auth,
+            "cookies": self.cookies,
+            "json": self.json,
+            "response_content_type": self.response_content_type
+        }
+
+
+class ForwardedResponse(BaseModel):
+    status_code: int | None = None
+    headers: Dict[str, Any] = dict[str, Any]()
+    url: str | None = None
+    encoding: str | None = None
+    reason: str | None = None
+    cookies: Dict[str, Any] = dict[str, Any]()
+    elapsed: datetime.timedelta = datetime.timedelta(0)
+    content: Any | None
+    content_type: ResponseType = 'text'
+
+    @staticmethod
+    def from_response(response: requests.Response, response_content_type: ResponseType = 'text') -> 'ForwardedResponse':
+        forwarded_response = ForwardedResponse(
+            status_code=response.status_code,
+            headers=response.headers,
+            url=response.url,
+            encoding=response.encoding,
+            reason=response.reason,
+            cookies=response.cookies,
+            elapsed=response.elapsed,
+            content=None,
+            content_type=response_content_type
+        )
+        match forwarded_response.content_type:
+            case 'raw':
+                forwarded_response.content = response.content
+            case 'text':
+                forwarded_response.content = response.text
+            case 'json':
+                forwarded_response.content = response.json()
+        return forwarded_response
+
+    @staticmethod
+    def from_json(json_dict: Dict[str, Any]) -> 'ForwardedResponse':
+        return ForwardedResponse(
+            status_code=json_dict['status_code'],
+            headers=json_dict['headers'],
+            url=json_dict['url'],
+            encoding=json_dict['encoding'],
+            reason=json_dict['reason'],
+            cookies=json_dict['cookies'],
+            elapsed=json_dict['elapsed'],
+            content=json_dict['content'],
+            content_type=json_dict['content_type'],
+        )
+
+    def to_json_dict(self):
+        def serialize_timedelta(timedelta: datetime.timedelta):
+            return {
+                'days': timedelta.days,
+                'seconds': timedelta.seconds,
+                'microseconds': timedelta.microseconds
+            }
+
+        return {
+            "status_code": self.status_code,
+            "headers": self.headers,
+            "url": self.url,
+            "encoding": self.encoding,
+            "reason": self.reason,
+            "cookies": self.cookies,
+            "elapsed": serialize_timedelta(self.elapsed),
+            "content": self.content,
+            "content_type": self.content_type,
+        }

requirements.txt

@@ -0,0 +1,6 @@
+uvicorn[standard]
+fastapi
+requests
+pydantic
+websocket-client
+wsproto