diff --git a/api/security.py b/api/security.py index 8ad8207..92308c7 100644 --- a/api/security.py +++ b/api/security.py @@ -1,3 +1,5 @@ +"""Security checks for the API. Checks if the IP is masked etc.""" + import os import httpx @@ -10,17 +12,6 @@ from dotenv import load_dotenv load_dotenv() is_proxy_enabled = False -def enable_proxy(): - """Enables the proxy.""" - - global is_proxy_enabled - - proxies.activate_proxy() - - print(f'[green]SUCCESS: Proxy enabled: {proxies.active_proxy}[/green]') - - is_proxy_enabled = True - class InsecureIPError(Exception): """Raised when the IP address of the server is not secure.""" diff --git a/api/tests.py b/api/tests.py new file mode 100644 index 0000000..fe4f6e0 --- /dev/null +++ b/api/tests.py @@ -0,0 +1,62 @@ +import os +import httpx +import proxies +import asyncio +import aiohttp +import aiohttp_socks + +from rich import print +from dotenv import load_dotenv + +load_dotenv() + +async def check_proxy(): + """Checks if the proxy is working.""" + + proxy = proxies.active_proxy + + connector = aiohttp_socks.ProxyConnector( + proxy_type=aiohttp_socks.ProxyType.SOCKS5, + host=proxy.ip_address, + port=proxy.port, + rdns=False, + username=proxy.username, + password=proxy.password + ) + + async with aiohttp.ClientSession(connector=connector) as session: + async with session.get('https://echo.hoppscotch.io/') as response: + json_data = await response.json() + return json_data['headers']['x-forwarded-for'] + +async def check_api(): + """Checks if the API is working.""" + + model = 'gpt-3.5-turbo' + messages = [ + { + 'role': 'user', + 'content': '2+2=' + }, + ] + + headers = { + 'Content-Type': 'application/json', + 'Authorization': 'Bearer ' + os.getenv('CLOSEDAI_KEY') + } + + json_data = { + 'model': model, + 'messages': messages, + 'stream': True + } + + async with httpx.AsyncClient(timeout=20) as client: + async with client.stream('POST', 'https://api.openai.com/v1/chat/completions', headers=headers, json=json_data) as response: + response.raise_for_status() + async for chunk in response.aiter_text(): + print(chunk.strip()) + +if __name__ == '__main__': + # asyncio.run(check_api()) + asyncio.run(check_proxy())