diff --git a/API/main.py b/API/main.py index 24bb15e..a9aade8 100644 --- a/API/main.py +++ b/API/main.py @@ -1,9 +1,9 @@ -from fastapi import FastAPI, HTTPException, status +from fastapi import FastAPI, HTTPException, Query, status from pydantic import BaseModel, EmailStr import utils app = FastAPI() -db = utils.DataBase("") +methods = utils.API("") class RegisterBody(BaseModel) : email : EmailStr @@ -13,10 +13,14 @@ class LoginBody(BaseModel) : email : EmailStr password : str +class UrlsBody(BaseModel) : + urls_name : str + email : EmailStr + password : str @app.post("/registration", status_code=status.HTTP_201_CREATED) def registration(body: RegisterBody) : - result = db.registration(body.email, body.password) + result = methods.registration(body.email, body.password) if result["code"] == 1: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, @@ -26,11 +30,59 @@ def registration(body: RegisterBody) : @app.post("/login", status_code=status.HTTP_200_OK) def login(body: LoginBody) : - answer = db.login(body.email, body.password) + answer = methods.login(body.email, body.password) if answer["code"] == 1 : raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect email or password." ) else : - return answer["data"] \ No newline at end of file + return answer["data"] + +@app.post("/add_url", status_code=status.HTTP_200_OK) +def add_url(body: UrlsBody) : + answer = methods.add_url(body.email, body.password, body.urls_name) + if answer["code"] == 1 : + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Incorrect email or password." + ) + elif answer["code"] == 2 : + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="The url already exists." + ) + else : + return answer["data"] + +@app.get("/get_url", status_code=status.HTTP_200_OK) +def get_url(email: EmailStr = Query(...), password: str = Query(...), urls_name: str = Query(...)) : + answer = methods.get_url(email, password, urls_name) + if answer["code"] == 1 : + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Incorrect email or password." + ) + elif answer["code"] == 2 : + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="The url does not exist" + ) + else : + return answer["data"] + +@app.delete("/del_url", status_code=status.HTTP_200_OK) +def del_url(email: EmailStr = Query(...), password: str = Query(...), urls_name: str = Query(...)) : + answer = methods.del_url(email, password, urls_name) + if answer["code"] == 1 : + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Incorrect email or password." + ) + elif answer["code"] == 2 : + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="The url does not exist" + ) + else : + return answer["data"] diff --git a/API/utils.py b/API/utils.py index f046c77..a5dc99a 100644 --- a/API/utils.py +++ b/API/utils.py @@ -1,36 +1,42 @@ import sqlite3 import pandas +import py3xui import datetime import hashlib import json +import uuid -class DataBase : - def __init__(self, path: str) : +class API : + def __init__(self, path: str, host: str, username: str, passwd: str, inbaund_id: int, inbaund_url: str, inbaund_port: str) : self.con = sqlite3.connect(path) self.cur = self.con.cursor() - self.cur.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, email TEXT NOT NULL UNIQUE, pass_hash TEXT NOT NULL, register TEXT NOT NULL, - urls TEXT DEFAULT "", + urls TEXT DEFAULT "[]" NOT NULL, payment TEXT DEFAULT "" ) """) - self.con.commit() + self.x_ui = py3xui.Api(host, username, passwd) + self.x_ui.login() + self.inbaund_id = inbaund_id + self.inbaund_url = inbaund_url + self.inbaund_port = inbaund_port + def read_table(self, sql_command: str, params=None) -> dict : if params == None : params = () return json.loads(pandas.read_sql(sql_command, self.con, params=params).to_json()) - def get_hash(text: str) : + def get_hash(text: str) -> str: return hashlib.sha256(text.encode()).hexdigest() - def registration(self, email: str, passwd: str) : + def registration(self, email: str, passwd: str) -> dict : """ 0 - Success 1 - Email is busy @@ -48,19 +54,112 @@ class DataBase : SELECT id FROM users WHERE email = ? """, (email)) - return {"code": 0, "data": {"id": data["id"]["0"], "email": email}} + return {"code": 0, "data": {"id": int(data["id"]["0"]), "email": email}} - def login(self, email: str, passwd: str) : + def login(self, email: str, passwd: str) -> dict : """ 0 - Success 1 - Incorrect email or password """ data = self.read_table(""" - SELECT id, urls, payment FROM users + SELECT id, urls FROM users WHERE email = ? AND pass_hash = ? """, (email, self.get_hash(passwd))) if data["id"] == {} : return {"code": 1} - else : - return {"code": 0, "data": {"id": data["id"]["0"], "email": email, "urls": data["urls"]["0"], "payment": data["payment"]["0"]}} + return {"code": 0, "data": {"id": int(data["id"]["0"]), "email": email, "urls": json.loads(data["urls"]["0"])}} + def add_url(self, email: str, passwd: str, url_name: str) -> dict : + """ + 0 - Success + 1 - Incorrect email or password + 2 - The url already exists + """ + data = self.read_table(""" + SELECT id, urls FROM users + WHERE email = ? AND pass_hash = ? + """, (email, self.get_hash(passwd))) + if data["id"] == {} : + return {"code": 1} + urls = json.loads(data["urls"]["0"]) + if url_name in urls : + return {"code": 2} + + uid = str(uuid.uuid4()) + new_client = py3xui.Client(id=uid, email=f"{email}-{url_name}", enable=True, flow="xtls-rprx-vision") + self.x_ui.client.add(self.inbaund_id, new_client) + + inbound = py3xui.Inbound(id=self.inbaund_id) + pbk = inbound.stream_settings.reality_settings.get("settings").get("publicKey") + wn = inbound.stream_settings.reality_settings.get("serverNames")[0] + short_id = inbound.stream_settings.reality_settings.get("shortIds")[0] + url = f"vless://{uid}@{self.inbaund_url}:{self.inbaund_port}?security=reality&pbk={pbk}&fp=random&sni={wn}&sid={short_id}&spx=%2F&flow=xtls-rprx-vision#SpectralVPN-{url_name}" + + urls.append(url_name) + + self.cur.execute(""" + UPDATE users SET + urls = ? + WHERE id = ? + """, (json.dumps(urls), int(data["id"]["0"]))) + data = self.read_table(""" + SELECT id, urls FROM users + WHERE email = ? AND pass_hash = ? + """, (email, self.get_hash(passwd))) + + return {"code": 0, "data": {"id": int(data["id"]["0"]), "email": email, "urls": json.loads(data["urls"]["0"]), "url": url}} + + def get_url(self, email: str, passwd: str, url_name: str) -> dict : + """ + 0 - Success + 1 - Incorrect email or password + 2 - The url does not exist + """ + data = self.read_table(""" + SELECT id, urls FROM users + WHERE email = ? AND pass_hash = ? + """, (email, self.get_hash(passwd))) + if data["id"] == {} : + return {"code": 1} + urls = json.loads(data["urls"]["0"]) + if not url_name in urls : + return {"code": 2} + + uid = self.x_ui.client.get_by_email(f"{email}-{url_name}").uuid + inbound = py3xui.Inbound(id=self.inbaund_id) + pbk = inbound.stream_settings.reality_settings.get("settings").get("publicKey") + wn = inbound.stream_settings.reality_settings.get("serverNames")[0] + short_id = inbound.stream_settings.reality_settings.get("shortIds")[0] + url = f"vless://{uid}@{self.inbaund_url}:{self.inbaund_port}?security=reality&pbk={pbk}&fp=random&sni={wn}&sid={short_id}&spx=%2F&flow=xtls-rprx-vision#SpectralVPN-{url_name}" + + return {"code": 0, "data": {"id": int(data["id"]["0"]), "email": email, "urls": urls, "url": url}} + + def del_url(self, email: str, passwd: str, url_name: str) -> dict : + """ + 0 - Success + 1 - Incorrect email or password + 2 - The url does not exist + """ + data = self.read_table(""" + SELECT id, urls FROM users + WHERE email = ? AND pass_hash = ? + """, (email, self.get_hash(passwd))) + if data["id"] == {} : + return {"code": 1} + urls = json.loads(data["urls"]["0"]) + if not url_name in urls : + return {"code": 2} + + uid = self.x_ui.client.get_by_email(f"{email}-{url_name}").uuid + self.x_ui.client.delete(self.inbaund_id, uid) + urls.remove(url_name) + self.cur.execute(""" + UPDATE users SET + urls = ? + WHERE id = ? + """, (json.dumps(urls), int(data["id"]["0"]))) + data = self.read_table(""" + SELECT id, urls FROM users + WHERE email = ? AND pass_hash = ? + """, (email, self.get_hash(passwd))) + return {"code": 0, "data": {"id": int(data["id"]["0"]), "email": email, "urls": urls}} \ No newline at end of file diff --git a/requirments.txt b/requirments.txt index ca166b7..ebd657d 100644 --- a/requirments.txt +++ b/requirments.txt @@ -1,2 +1,3 @@ fastapi[all] -pandas \ No newline at end of file +pandas +py3xui \ No newline at end of file