236 lines
8.7 KiB
Python
236 lines
8.7 KiB
Python
import aiohttp
|
|
import uuid
|
|
import json
|
|
import base64
|
|
import ssl
|
|
from datetime import datetime
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
def generate_uuid():
|
|
return str(uuid.uuid4())
|
|
|
|
|
|
class PanelInteraction:
|
|
def __init__(self, base_url, login_data, logger, certificate=None, is_encoded=True):
|
|
"""
|
|
Initialize the PanelInteraction class.
|
|
|
|
:param base_url: Base URL for the panel.
|
|
:param login_data: Login data (username/password or token).
|
|
:param logger: Logger for debugging.
|
|
:param certificate: Certificate content (Base64-encoded or raw string).
|
|
:param is_encoded: Indicates whether the certificate is Base64-encoded.
|
|
"""
|
|
self.base_url = base_url
|
|
self.login_data = login_data
|
|
self.logger = logger
|
|
self.cert_content = self._decode_certificate(certificate, is_encoded)
|
|
self.session_id = None # Session ID will be initialized lazily
|
|
self.headers = None
|
|
|
|
def _decode_certificate(self, certificate, is_encoded):
|
|
"""
|
|
Decode the provided certificate content.
|
|
|
|
:param certificate: Certificate content (Base64-encoded or raw string).
|
|
:param is_encoded: Indicates whether the certificate is Base64-encoded.
|
|
:return: Decoded certificate content as bytes.
|
|
"""
|
|
|
|
if not certificate:
|
|
self.logger.error("No certificate provided.")
|
|
raise ValueError("Certificate is required.")
|
|
try:
|
|
# Создаем SSLContext
|
|
ssl_context = ssl.create_default_context()
|
|
|
|
# Декодируем, если нужно
|
|
if is_encoded:
|
|
certificate = base64.b64decode(certificate).decode()
|
|
|
|
# Загружаем сертификат в SSLContext
|
|
ssl_context.load_verify_locations(cadata=certificate)
|
|
return ssl_context
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error while decoding certificate: {e}")
|
|
raise ValueError("Invalid certificate format or content.") from e
|
|
|
|
|
|
async def _ensure_logged_in(self):
|
|
"""
|
|
Ensure the session ID is available for authenticated requests.
|
|
"""
|
|
if not self.session_id:
|
|
self.session_id = await self.login()
|
|
if self.session_id:
|
|
self.headers = {
|
|
'Accept': 'application/json',
|
|
'Cookie': f'3x-ui={self.session_id}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
else:
|
|
raise ValueError("Unable to log in and retrieve session ID.")
|
|
|
|
async def login(self):
|
|
"""
|
|
Perform login to the panel.
|
|
|
|
:return: Session ID or None.
|
|
"""
|
|
login_url = f"{self.base_url}/login"
|
|
self.logger.info(f"Attempting to login at: {login_url}")
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
try:
|
|
async with session.post(
|
|
login_url, data=self.login_data, ssl=self.cert_content, timeout=10
|
|
) as response:
|
|
if response.status == 200:
|
|
session_id = response.cookies.get("3x-ui")
|
|
if session_id:
|
|
return session_id.value
|
|
else:
|
|
self.logger.error("Login failed: No session ID received.")
|
|
return None
|
|
else:
|
|
self.logger.error(f"Login failed: {response.status}")
|
|
return None
|
|
except aiohttp.ClientError as e:
|
|
self.logger.error(f"Login request failed: {e}")
|
|
return None
|
|
|
|
async def get_inbound_info(self, inbound_id):
|
|
"""
|
|
Fetch inbound information by ID.
|
|
|
|
:param inbound_id: ID of the inbound.
|
|
:return: JSON response or None.
|
|
"""
|
|
await self._ensure_logged_in()
|
|
url = f"{self.base_url}/panel/api/inbounds/get/{inbound_id}"
|
|
async with aiohttp.ClientSession() as session:
|
|
try:
|
|
async with session.get(
|
|
url, headers=self.headers, ssl=self.cert_content, timeout=10
|
|
) as response:
|
|
if response.status == 200:
|
|
return await response.json()
|
|
else:
|
|
self.logger.error(f"Failed to get inbound info: {response.status}")
|
|
return None
|
|
except aiohttp.ClientError as e:
|
|
self.logger.error(f"Get inbound info request failed: {e}")
|
|
return None
|
|
|
|
async def get_client_traffic(self, email):
|
|
"""
|
|
Fetch traffic information for a specific client.
|
|
|
|
:param email: Client's email.
|
|
:return: JSON response or None.
|
|
"""
|
|
await self._ensure_logged_in()
|
|
url = f"{self.base_url}/panel/api/inbounds/getClientTraffics/{email}"
|
|
async with aiohttp.ClientSession() as session:
|
|
try:
|
|
async with session.get(
|
|
url, headers=self.headers, ssl=self.cert_content, timeout=10
|
|
) as response:
|
|
if response.status == 200:
|
|
return await response.json()
|
|
else:
|
|
self.logger.error(f"Failed to get client traffic: {response.status}")
|
|
return None
|
|
except aiohttp.ClientError as e:
|
|
self.logger.error(f"Get client traffic request failed: {e}")
|
|
return None
|
|
|
|
async def update_client_expiry(self, client_uuid, new_expiry_time, client_email):
|
|
"""
|
|
Update the expiry date of a specific client.
|
|
|
|
:param client_uuid: UUID of the client.
|
|
:param new_expiry_time: New expiry date in ISO format.
|
|
:param client_email: Client's email.
|
|
:return: None.
|
|
"""
|
|
await self._ensure_logged_in()
|
|
url = f"{self.base_url}/panel/api/inbounds/updateClient"
|
|
update_data = {
|
|
"id": 1,
|
|
"settings": {
|
|
"clients": [
|
|
{
|
|
"id": client_uuid,
|
|
"alterId": 0,
|
|
"email": client_email,
|
|
"limitIp": 2,
|
|
"totalGB": 0,
|
|
"expiryTime": new_expiry_time,
|
|
"enable": True,
|
|
"tgId": "",
|
|
"subId": ""
|
|
}
|
|
]
|
|
}
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
try:
|
|
async with session.post(
|
|
url, headers=self.headers, json=update_data, ssl=self.cert_content
|
|
) as response:
|
|
if response.status == 200:
|
|
self.logger.info("Client expiry updated successfully.")
|
|
else:
|
|
self.logger.error(f"Failed to update client expiry: {response.status}")
|
|
except aiohttp.ClientError as e:
|
|
self.logger.error(f"Update client expiry request failed: {e}")
|
|
|
|
async def add_client(self, inbound_id, expiry_date, email):
|
|
"""
|
|
Add a new client to an inbound.
|
|
|
|
:param inbound_id: ID of the inbound.
|
|
:param expiry_date: Expiry date in ISO format.
|
|
:param email: Client's email.
|
|
:return: JSON response or None.
|
|
"""
|
|
await self._ensure_logged_in()
|
|
url = f"{self.base_url}/panel/api/inbounds/addClient"
|
|
client_info = {
|
|
"clients": [
|
|
{
|
|
"id": generate_uuid(),
|
|
"alterId": 0,
|
|
"email": email,
|
|
"limitIp": 2,
|
|
"totalGB": 0,
|
|
"flow": "xtls-rprx-vision",
|
|
"expiryTime": expiry_date,
|
|
"enable": True,
|
|
"tgId": "",
|
|
"subId": ""
|
|
}
|
|
]
|
|
}
|
|
payload = {
|
|
"id": inbound_id,
|
|
"settings": client_info
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
try:
|
|
async with session.post(
|
|
url, headers=self.headers, json=payload, ssl=self.cert_content
|
|
) as response:
|
|
if response.status == 200:
|
|
return await response.status
|
|
else:
|
|
self.logger.error(f"Failed to add client: {response.status}")
|
|
return None
|
|
except aiohttp.ClientError as e:
|
|
self.logger.error(f"Add client request failed: {e}")
|
|
return None
|