diff --git a/app/__init__.py b/app/__init__.py index 96ebe15..2470d6a 100755 --- a/app/__init__.py +++ b/app/__init__.py @@ -1,72 +1,18 @@ import logging -from apscheduler.schedulers.asyncio import AsyncIOScheduler -from fastapi import FastAPI, Request, status -from fastapi.encoders import jsonable_encoder -from fastapi.exceptions import RequestValidationError -from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import JSONResponse -from fastapi.routing import APIRoute -from fastapi_responses import custom_openapi +import uvicorn -from config import DOCS, XRAY_SUBSCRIPTION_PATH +import config -__version__ = "0.4.2" +__version__ = "0.1.0" - -app = FastAPI( - title="MarzbanAPI", - description="Unified GUI Censorship Resistant Solution Powered by Xray", - version=__version__, - docs_url="/docs" if DOCS else None, - redoc_url="/redoc" if DOCS else None, -) -app.openapi = custom_openapi(app) -scheduler = AsyncIOScheduler(timezone="UTC") -logger = logging.getLogger("uvicorn.error") -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], +logger = logging.getLogger(__name__) +logger.setLevel(logging.DEBUG if config.DEBUG else logging.INFO) +handler = logging.StreamHandler() +formatter = uvicorn.logging.ColourizedFormatter( + "{levelprefix:<8} @{name}: {message}", + style="{", + use_colors=True ) - - -from app import dashboard, views, jobs # , telegram # noqa - - -def use_route_names_as_operation_ids(app: FastAPI) -> None: - for route in app.routes: - if isinstance(route, APIRoute): - route.operation_id = route.name - - -use_route_names_as_operation_ids(app) - - -@app.on_event("startup") -def on_startup(): - paths = [f"{r.path}/" for r in app.routes] - paths.append("/api/") - if f"/{XRAY_SUBSCRIPTION_PATH}/" in paths: - raise ValueError( - f"you can't use /{XRAY_SUBSCRIPTION_PATH}/ as subscription path it reserved for {app.title}" - ) - scheduler.start() - - -@app.on_event("shutdown") -def on_shutdown(): - scheduler.shutdown() - - -@app.exception_handler(RequestValidationError) -def validation_exception_handler(request: Request, exc: RequestValidationError): - details = {} - for error in exc.errors(): - details[error["loc"][-1]] = error.get("msg") - return JSONResponse( - status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, - content=jsonable_encoder({"detail": details}), - ) +handler.setFormatter(formatter) +logger.addHandler(handler) diff --git a/app/dashboard/__init__.py b/app/dashboard/__init__.py index 8138e42..b49a13b 100644 --- a/app/dashboard/__init__.py +++ b/app/dashboard/__init__.py @@ -1,11 +1,8 @@ -import atexit import os import subprocess from pathlib import Path -from app import app from config import DEBUG, VITE_BASE_API -from fastapi.staticfiles import StaticFiles path = '/dashboard/' base_dir = Path(__file__).parent @@ -22,33 +19,4 @@ def build(): with open(build_dir / 'index.html', 'r') as file: html = file.read() with open(build_dir / '404.html', 'w') as file: - file.write(html) - - -def run_dev(): - proc = subprocess.Popen( - ['npm', 'run', 'dev', '--', '--host', '0.0.0.0', '--base', path, '--clearScreen', 'false'], - env={**os.environ, 'VITE_BASE_API': VITE_BASE_API}, - cwd=base_dir - ) - - atexit.register(proc.terminate) - - -def run_build(): - if not build_dir.is_dir(): - build() - - app.mount( - path, - StaticFiles(directory=build_dir, html=True), - name="dashboard" - ) - - -@app.on_event("startup") -def startup(): - if DEBUG is True: - run_dev() - else: - run_build() + file.write(html) \ No newline at end of file diff --git a/app/db/__init__.py b/app/db/__init__.py index 67880dc..1f18c50 100644 --- a/app/db/__init__.py +++ b/app/db/__init__.py @@ -1,10 +1,19 @@ from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm import Session -from sqlalchemy.schema import MetaData from .base import Base, SessionLocal, engine # noqa -# meta_obj = MetaData() +from .crud import (create_admin, create_notification_reminder, # noqa + create_user, delete_notification_reminder, get_admin, + get_admins, get_jwt_secret_key, get_notification_reminder, + assure_node_inbounds, get_system_usage, + get_tls_certificate, get_user, get_user_by_id, get_users, + get_users_count, remove_admin, remove_user, revoke_user_sub, + set_owner, update_admin, update_user, update_user_status, + update_user_sub, start_user_expire) + +from .models import JWT, System, User # noqa + class GetDB: # Context Manager def __init__(self): @@ -20,26 +29,7 @@ class GetDB: # Context Manager self.db.close() -def get_db(): # Dependency - with GetDB() as db: - yield db - - -from .crud import (create_admin, create_notification_reminder, # noqa - create_user, delete_notification_reminder, get_admin, - get_admins, get_jwt_secret_key, get_notification_reminder, - assure_node_inbounds, get_system_usage, - get_tls_certificate, get_user, get_user_by_id, get_users, - get_users_count, remove_admin, remove_user, revoke_user_sub, - set_owner, update_admin, update_user, update_user_status, - update_user_sub, start_user_expire) - -from .models import JWT, System, User # noqa - -# meta_obj.create_all(engine) - __all__ = [ - "get_or_create_inbound", "get_user", "get_user_by_id", "get_users", @@ -66,7 +56,6 @@ __all__ = [ "delete_notification_reminder", "GetDB", - "get_db", "User", "System", diff --git a/app/db/crud.py b/app/db/crud.py index a406a62..fcf8cd7 100644 --- a/app/db/crud.py +++ b/app/db/crud.py @@ -19,7 +19,7 @@ from app.models.proxy import InboundHost as InboundHostModify from app.models.user import (ReminderType, UserCreate, UserDataLimitResetStrategy, UserModify, UserResponse, UserStatus, UserUsageResponse) -from app.models.service import Service as ServiceModify +from app.models.service import Service as ServiceModify, ServiceCreate # from app.models.user_template import UserTemplateCreate, UserTemplateModify from app.utils.helpers import (calculate_expiration_days, calculate_usage_percent) @@ -482,10 +482,7 @@ def get_admins(db: Session, return query.all() -def create_service(db: Session, service: Service) -> Service: - - #for _, i in service.inbounds.items(): - # inbound_tags.extend(i) +def create_service(db: Session, service: ServiceCreate) -> Service: dbservice = Service( name=service.name, inbounds=db.query(Inbound).filter(Inbound.id.in_(service.inbounds)).all(), @@ -497,14 +494,32 @@ def create_service(db: Session, service: Service) -> Service: return dbservice -def get_service(db: Session, service_name: str) -> Service: - return db.query(Service).filter( Service.name == service_name ).first() +def get_service(db: Session, service_id: id) -> Service: + return db.query(Service).filter(Service.id == service_id).first() def get_services(db: Session) -> List[Service]: return db.query(Service).all() +def update_service(db: Session, db_service: Service, modification: ServiceModify): + if modification.name is not None: + db_service.name = modification.name + + if modification.inbounds is not None: + db_service.inbounds = db.query(Inbound).filter(Inbound.id.in_(modification.inbounds)).all() + + db.commit() + db.refresh(db_service) + return db_service + + +def remove_service(db: Session, db_service: Service): + db.delete(db_service) + db.commit() + return db_service + + def get_node(db: Session, name: str): return db.query(Node).filter(Node.name == name).first() diff --git a/app/dependencies.py b/app/dependencies.py new file mode 100644 index 0000000..469de6d --- /dev/null +++ b/app/dependencies.py @@ -0,0 +1,106 @@ +from datetime import datetime +from typing import Annotated + +from fastapi import Depends, HTTPException, status +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session + +from app.db import crud, User, GetDB +from app.models.admin import Admin, oauth2_scheme +from app.utils.jwt import get_admin_payload +from config import SUDOERS + + +def get_db(): # Dependency + with GetDB() as db: + yield db + + +def get_admin(db: Annotated[Session, Depends(get_db)], + token: Annotated[str, Depends(oauth2_scheme)]): + payload = get_admin_payload(token) + if not payload: + return + + if payload['username'] in SUDOERS and payload['is_sudo'] is True: + return Admin(username=payload['username'], is_sudo=True) + + dbadmin = crud.get_admin(db, payload['username']) + if not dbadmin: + return + + if dbadmin.password_reset_at: + if not payload.get("created_at"): + return + if dbadmin.password_reset_at > payload.get("created_at"): + return + + return Admin.model_validate(dbadmin) + + +def get_current_admin(admin: Annotated[Admin, Depends(get_admin)]): + if not admin: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Could not validate credentials", + headers={"WWW-Authenticate": "Bearer"}, + ) + + return admin + + +def sudo_admin(admin: Annotated[Admin, Depends(get_current_admin)]): + if not admin.is_sudo: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Access Denied", + ) + return admin + + +def get_subscription_user(username: str, key: str, db: Annotated[Session, Depends(get_db)]): + try: + int(key, 16) + except ValueError: + raise HTTPException(404) + + db_user = crud.get_user(db, username) + if not db_user or db_user.key != key: + raise HTTPException(404) + return db_user + + +def get_user(username: str, + admin: Annotated[Admin, Depends(get_current_admin)], + db: Annotated[Session, Depends(get_db)]): + db_user = crud.get_user(db, username) + if not (admin.is_sudo or (db_user.admin and db_user.admin.username == admin.username)): + raise HTTPException(status_code=403, detail="You're not allowed") + + if not db_user: + raise HTTPException(status_code=404, detail="User not found") + + return db_user + + +def parse_start_date(start: str | None = None): + if not start: + return datetime.fromtimestamp(datetime.utcnow().timestamp() - 30 * 24 * 3600) + else: + return datetime.fromisoformat(start) + + +def parse_end_date(end: str | None = None): + if not end: + return datetime.utcnow() + else: + return datetime.fromisoformat(end) + + +SubUserDep = Annotated[User, Depends(get_subscription_user)] +UserDep = Annotated[User, Depends(get_user)] +AdminDep = Annotated[Admin, Depends(get_current_admin)] +SudoAdminDep = Annotated[Admin, Depends(sudo_admin)] +DBDep = Annotated[Session, Depends(get_db)] +StartDateDep = Annotated[datetime, Depends(parse_start_date)] +EndDateDep = Annotated[datetime, Depends(parse_end_date)] diff --git a/app/jobs/__init__.py b/app/jobs/__init__.py deleted file mode 100644 index 627a8bd..0000000 --- a/app/jobs/__init__.py +++ /dev/null @@ -1,13 +0,0 @@ -import glob -import importlib.util -from os.path import basename, dirname, join - -modules = glob.glob(join(dirname(__file__), "*.py")) - -for file in modules: - name = basename(file).replace('.py', '') - if name.startswith('_'): - continue - - spec = importlib.util.spec_from_file_location(name, file) - spec.loader.exec_module(importlib.util.module_from_spec(spec)) diff --git a/app/marzneshin.py b/app/marzneshin.py new file mode 100644 index 0000000..a15236a --- /dev/null +++ b/app/marzneshin.py @@ -0,0 +1,124 @@ +import asyncio +import atexit +import logging +import os +from datetime import datetime as dt +from datetime import timedelta as td + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from fastapi import FastAPI, Request, status +from fastapi.encoders import jsonable_encoder +from fastapi.exceptions import RequestValidationError +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import HTMLResponse +from fastapi.responses import JSONResponse +from starlette.staticfiles import StaticFiles +from uvicorn import Server, Config + +import config +from app.templates import render_template +from config import DOCS, DEBUG, UVICORN_SSL_KEYFILE, UVICORN_SSL_CERTFILE, UVICORN_UDS, UVICORN_PORT, UVICORN_HOST +from config import HOME_PAGE_TEMPLATE +from . import __version__, telegram +from .dashboard import build_dir, build, base_dir +from .tasks import record_user_usages, review_users, reset_user_data_usage, nodes_health_check, \ + record_realtime_bandwidth, send_notifications, delete_expired_reminders, nodes_startup +from .routes import admin, node, service, subscription, system, user + +logger = logging.getLogger(__name__) +dashboard_path = "/dashboard/" + +app = FastAPI( + title="MarzneshinAPI", + description="Unified GUI Censorship Resistant Solution Powered by Xray", + version=__version__, + docs_url="/docs" if DOCS else None, + redoc_url="/redoc" if DOCS else None, +) + +app.include_router(admin.router) +app.include_router(node.router) +app.include_router(service.router) +app.include_router(subscription.router) +app.include_router(system.router) +app.include_router(user.router) + + +@app.get("/", response_class=HTMLResponse) +def home(): + return render_template(HOME_PAGE_TEMPLATE) + + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +scheduler = AsyncIOScheduler(timezone="UTC") +scheduler.add_job(record_user_usages, 'interval', coalesce=True, seconds=30) +scheduler.add_job(review_users, 'interval', seconds=600, coalesce=True, max_instances=1) +scheduler.add_job(reset_user_data_usage, 'interval', coalesce=True, hours=1) +scheduler.add_job(nodes_health_check, 'interval', seconds=5, coalesce=True, max_instances=1) +scheduler.add_job(record_realtime_bandwidth, 'interval', seconds=1, coalesce=True, max_instances=1) + +if config.WEBHOOK_ADDRESS: + scheduler.add_job(send_notifications, "interval", seconds=30, replace_existing=True) + scheduler.add_job(delete_expired_reminders, "interval", hours=2, start_date=dt.utcnow() + td(minutes=1)) + + +@app.on_event("shutdown") +async def on_shutdown(): + scheduler.shutdown() + logger.info("Sending pending notifications before shutdown...") + await send_notifications() + + +@app.exception_handler(RequestValidationError) +def validation_exception_handler(request: Request, exc: RequestValidationError): + details = {} + for error in exc.errors(): + details[error["loc"][-1]] = error.get("msg") + return JSONResponse( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + content=jsonable_encoder({"detail": details}), + ) + + +async def main(): + if DEBUG: + proc = await asyncio.create_subprocess_exec( + 'npm', + 'run', 'dev', '--', '--host', '0.0.0.0', '--base', + dashboard_path, '--clearScreen', 'false', + env={**os.environ, 'VITE_BASE_API': config.VITE_BASE_API}, + cwd=base_dir + ) + atexit.register(proc.terminate) + else: + if not build_dir.is_dir(): + build() + + app.mount( + "/dashboard/", + StaticFiles(directory=build_dir, html=True), + name="dashboard" + ) + scheduler.start() + asyncio.create_task(telegram.start_bot()) + await nodes_startup() + cfg = Config( + app=app, + host=UVICORN_HOST, + port=UVICORN_PORT, + uds=(None if DEBUG else UVICORN_UDS), + ssl_certfile=UVICORN_SSL_CERTFILE, + ssl_keyfile=UVICORN_SSL_KEYFILE, + workers=1, + reload=DEBUG, + log_level=logging.DEBUG if DEBUG else logging.INFO, + ) + server = Server(cfg) + await server.serve() diff --git a/app/models/admin.py b/app/models/admin.py index 2df71db..41e9982 100644 --- a/app/models/admin.py +++ b/app/models/admin.py @@ -1,14 +1,9 @@ from typing import Optional -from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from passlib.context import CryptContext from pydantic import ConfigDict, BaseModel -from app.db import Session, crud, get_db -from app.utils.jwt import get_admin_payload -from config import SUDOERS - pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/admin/token") # Admin view url @@ -23,41 +18,6 @@ class Admin(BaseModel): is_sudo: bool model_config = ConfigDict(from_attributes=True) - @classmethod - def get_admin(cls, token: str, db: Session): - payload = get_admin_payload(token) - if not payload: - return - - if payload['username'] in SUDOERS and payload['is_sudo'] is True: - return cls(username=payload['username'], is_sudo=True) - - dbadmin = crud.get_admin(db, payload['username']) - if not dbadmin: - return - - if dbadmin.password_reset_at: - if not payload.get("created_at"): - return - if dbadmin.password_reset_at > payload.get("created_at"): - return - - return cls.from_orm(dbadmin) - - @classmethod - def get_current(cls, - db: Session = Depends(get_db), - token: str = Depends(oauth2_scheme)): - admin = cls.get_admin(token, db) - if not admin: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Could not validate credentials", - headers={"WWW-Authenticate": "Bearer"}, - ) - - return admin - class AdminCreate(Admin): password: str diff --git a/app/models/service.py b/app/models/service.py index 0e1b95a..7919aed 100644 --- a/app/models/service.py +++ b/app/models/service.py @@ -1,8 +1,6 @@ -from typing import Dict, List, Optional, Union +from typing import List, Optional -from pydantic import ConfigDict, BaseModel, Field, validator, computed_field - -# from app import xray +from pydantic import ConfigDict, BaseModel, Field, computed_field class ServiceBase(BaseModel): @@ -20,7 +18,7 @@ class Service(ServiceBase): inbounds: List[InboundBase] = Field([]) -class ServiceCreate(Service): +class ServiceCreate(ServiceBase): users: List[int] = Field([]) inbounds: List[int] = Field([]) model_config = ConfigDict(json_schema_extra={ @@ -32,22 +30,19 @@ class ServiceCreate(Service): }) -class ServiceModify(Service): - id: int +class ServiceModify(ServiceBase): + inbounds: Optional[List[int]] = Field(None) model_config = ConfigDict(json_schema_extra={ "example": { "id": 2, "name": "my service 2", - "inbounds": [1, 2, 3], - "users": [1, 2, 3], + "inbounds": [1, 2, 3] } }) class ServiceResponse(Service): id: int - users: List[UserBase] = Field([]) - inbounds: List[InboundBase] = Field([]) @computed_field @property @@ -59,18 +54,4 @@ class ServiceResponse(Service): def inbound_ids(self) -> List[int]: return [i.id for i in self.inbounds] - """ - @validator("inbounds", pre=True) - def validate_inbounds(cls, v): - final = {} - inbound_tags = [i.tag for i in v] - for protocol, inbounds in xray.config.inbounds_by_protocol.items(): - for inbound in inbounds: - if inbound["tag"] in inbound_tags: - if protocol in final: - final[protocol].append(inbound["tag"]) - else: - final[protocol] = [inbound["tag"]] - return final - """ model_config = ConfigDict(from_attributes=True) diff --git a/app/models/user.py b/app/models/user.py index 76ea2df..34d4a7f 100644 --- a/app/models/user.py +++ b/app/models/user.py @@ -7,7 +7,7 @@ from typing import List, Optional, Union from pydantic import field_validator, ConfigDict, BaseModel, Field, validator, computed_field, ValidationInfo from app.models.proxy import InboundBase -from config import XRAY_SUBSCRIPTION_PATH, XRAY_SUBSCRIPTION_URL_PREFIX +from config import XRAY_SUBSCRIPTION_URL_PREFIX USERNAME_REGEXP = re.compile(r"^(?=\w{3,32}\b)[a-zA-Z0-9-_@.]+(?:_[a-zA-Z0-9-_@.]+)*$") @@ -25,7 +25,7 @@ class UserStatus(str, Enum): on_hold = "on_hold" -from app.utils.share import generate_v2ray_links +# from app.utils.share import generate_v2ray_links class UserStatusModify(str, Enum): @@ -67,7 +67,7 @@ class UserBase(BaseModel): on_hold_expire_duration: Optional[int] = Field(None, nullable=True) on_hold_timeout: Optional[Union[datetime, None]] = Field(None, nullable=True) - + @field_validator("username") @classmethod def validate_username(cls, v: str): @@ -180,8 +180,8 @@ class UserResponse(User): created_at: datetime inbounds: List[InboundBase] = Field([]) services: List[ServiceBase] = Field([]) + links: List[str] | None = [] # Field(None) - # links: List[str] = [] # Field(default_factory=lamba x: ) # subscription_url: str = "" model_config = ConfigDict(from_attributes=True) @@ -190,19 +190,12 @@ class UserResponse(User): def service_ids(self) -> List[int]: return [s.id for s in self.services] - @computed_field - @property - def links(self) -> List[str]: - return generate_v2ray_links( - inbounds=self.inbounds, key=self.key, extra_data=self.dict(exclude={'subscription_url', 'links'}) - ) - @computed_field @property def subscription_url(self) -> str: salt = secrets.token_hex(8) url_prefix = (XRAY_SUBSCRIPTION_URL_PREFIX).replace('*', salt) - return f"{url_prefix}/{XRAY_SUBSCRIPTION_PATH}/{self.username}/{self.key}" + return f"{url_prefix}/sub/{self.username}/{self.key}" class UsersResponse(BaseModel): diff --git a/app/routes/__init__.py b/app/routes/__init__.py new file mode 100644 index 0000000..3ee40bc --- /dev/null +++ b/app/routes/__init__.py @@ -0,0 +1,12 @@ +import logging + +logger = logging.getLogger(__name__) + +__all__ = [ + "admin", + "node", + "service", + "subscription", + "system", + "user" +] diff --git a/app/views/admin.py b/app/routes/admin.py similarity index 61% rename from app/views/admin.py rename to app/routes/admin.py index fce4cee..1297e96 100644 --- a/app/views/admin.py +++ b/app/routes/admin.py @@ -1,13 +1,17 @@ -from typing import List, Optional +from typing import List, Optional, Annotated import sqlalchemy -from app import app -from app.db import Session, crud, get_db +from fastapi import APIRouter +from fastapi import Depends, HTTPException, status +from fastapi.security import OAuth2PasswordRequestForm + +from app.db import Session, crud +from app.dependencies import AdminDep, SudoAdminDep, DBDep from app.models.admin import Admin, AdminCreate, AdminInDB, AdminModify, Token from app.utils.jwt import create_admin_token from config import SUDOERS -from fastapi import Depends, HTTPException, status -from fastapi.security import OAuth2PasswordRequestForm + +router = APIRouter(tags=["Admin"]) def authenticate_env_sudo(username: str, password: str) -> bool: @@ -25,9 +29,9 @@ def authenticate_admin(db: Session, username: str, password: str) -> Optional[Ad return dbadmin if AdminInDB.model_validate(dbadmin).verify_password(password) else None -@app.post("/api/admin/token", tags=['Admin'], response_model=Token) -def admin_token(form_data: OAuth2PasswordRequestForm = Depends(), - db: Session = Depends(get_db)): +@router.post("/api/admin/token", response_model=Token) +def admin_token(form_data: Annotated[OAuth2PasswordRequestForm, Depends()], + db: DBDep): if authenticate_env_sudo(form_data.username, form_data.password): return Token(access_token=create_admin_token(form_data.username, is_sudo=True)) @@ -41,14 +45,10 @@ def admin_token(form_data: OAuth2PasswordRequestForm = Depends(), ) -@app.post("/api/admin", tags=['Admin'], response_model=Admin) +@router.post("/api/admin", response_model=Admin) def create_admin(new_admin: AdminCreate, - db: Session = Depends(get_db), - admin: Admin = Depends(Admin.get_current)): - - if not admin.is_sudo: - raise HTTPException(status_code=403, detail="You're not allowed") - + db: DBDep, + admin: SudoAdminDep): try: dbadmin = crud.create_admin(db, new_admin) except sqlalchemy.exc.IntegrityError: @@ -58,11 +58,11 @@ def create_admin(new_admin: AdminCreate, return dbadmin -@app.put("/api/admin/{username}", tags=['Admin'], response_model=Admin) +@router.put("/api/admin/{username}", response_model=Admin) def modify_admin(username: str, modified_admin: AdminModify, - db: Session = Depends(get_db), - admin: Admin = Depends(Admin.get_current)): + db: DBDep, + admin: AdminDep): if not (admin.is_sudo or admin.username == username): raise HTTPException(status_code=403, detail="You're not allowed") @@ -78,20 +78,17 @@ def modify_admin(username: str, if (username != admin.username) and dbadmin.is_sudo: raise HTTPException( status_code=403, - detail=("You're not allowed to edit another sudoers account. Use marzban-cli instead.",), + detail="You're not allowed to edit another sudoers account. Use marzban-cli instead.", ) dbadmin = crud.update_admin(db, dbadmin, modified_admin) return dbadmin -@app.delete("/api/admin/{username}", tags=['Admin']) +@router.delete("/api/admin/{username}") def remove_admin(username: str, - db: Session = Depends(get_db), - admin: Admin = Depends(Admin.get_current)): - if not admin.is_sudo: - raise HTTPException(status_code=403, detail="You're not allowed") - + db: DBDep, + admin: SudoAdminDep): dbadmin = crud.get_admin(db, username) if not dbadmin: raise HTTPException(status_code=404, detail="Admin not found") @@ -99,26 +96,22 @@ def remove_admin(username: str, if dbadmin.is_sudo: raise HTTPException( status_code=403, - detail=("You're not allowed to delete sudoers accounts. Use marzban-cli instead."), + detail="You're not allowed to delete sudoers accounts. Use marzban-cli instead.", ) - dbadmin = crud.remove_admin(db, dbadmin) + crud.remove_admin(db, dbadmin) return {} -@app.get("/api/admin", tags=["Admin"], response_model=Admin) -def get_current_admin(admin: Admin = Depends(Admin.get_current)): +@router.get("/api/admin", response_model=Admin) +def get_current_admin(admin: AdminDep): return admin -@app.get("/api/admins", tags=['Admin'], response_model=List[Admin]) -def get_admins(offset: int = None, +@router.get("/api/admins", response_model=List[Admin]) +def get_admins(db: DBDep, + admin: SudoAdminDep, + offset: int = None, limit: int = None, - username: str = None, - db: Session = Depends(get_db), - admin: Admin = Depends(Admin.get_current)): - - if not admin.is_sudo: - raise HTTPException(status_code=403, detail="You're not allowed") - + username: str = None): return crud.get_admins(db, offset, limit, username) diff --git a/app/views/node.py b/app/routes/node.py similarity index 51% rename from app/views/node.py rename to app/routes/node.py index 9220fff..070111b 100644 --- a/app/views/node.py +++ b/app/routes/node.py @@ -1,26 +1,26 @@ -import asyncio -# import time +import logging from datetime import datetime from typing import List import sqlalchemy -from fastapi import BackgroundTasks, Depends, HTTPException, WebSocket +from fastapi import APIRouter, Depends +from fastapi import HTTPException, WebSocket - -from app import app, logger, marznode -from app.db import Session, crud, get_db, get_tls_certificate +from app import marznode +from app.db import crud, get_tls_certificate +from app.dependencies import DBDep, SudoAdminDep, sudo_admin, EndDateDep, StartDateDep from app.marznode import MarzNodeGRPC from app.models.admin import Admin from app.models.node import (NodeCreate, NodeModify, NodeResponse, NodeSettings, NodeStatus, NodesUsageResponse) +logger = logging.getLogger(__name__) +router = APIRouter(tags=['Node'], dependencies=[Depends(sudo_admin)]) -@app.get("/api/node/settings", tags=['Node'], response_model=NodeSettings) -def get_node_settings(db: Session = Depends(get_db), - admin: Admin = Depends(Admin.get_current)): - if not admin.is_sudo: - raise HTTPException(status_code=403, detail="You're not allowed") +@router.get("/api/node/settings", response_model=NodeSettings) +def get_node_settings(db: DBDep, + admin: SudoAdminDep): tls = crud.get_tls_certificate(db) return NodeSettings( @@ -28,14 +28,10 @@ def get_node_settings(db: Session = Depends(get_db), ) -@app.post("/api/node", tags=['Node'], response_model=NodeResponse) +@router.post("/api/node", response_model=NodeResponse) async def add_node(new_node: NodeCreate, - db: Session = Depends(get_db), - admin: Admin = Depends(Admin.get_current)): - - if not admin.is_sudo: - raise HTTPException(status_code=403, detail="You're not allowed") - + db: DBDep, + admin: SudoAdminDep): try: db_node = crud.create_node(db, new_node) except sqlalchemy.exc.IntegrityError: @@ -48,26 +44,25 @@ async def add_node(new_node: NodeCreate, await marznode.operations.add_node(db_node.id, node) - logger.info(f"New node \"{db_node.name}\" added") + logger.info("New node `%s` added", db_node.name) return db_node -@app.get("/api/node/{node_id}", tags=['Node'], response_model=NodeResponse) +@router.get("/api/node/{node_id}", response_model=NodeResponse) def get_node(node_id: int, - db: Session = Depends(get_db), - admin: Admin = Depends(Admin.get_current)): - if not admin.is_sudo: - raise HTTPException(status_code=403, detail="You're not allowed") - - dbnode = crud.get_node_by_id(db, node_id) - if not dbnode: + db: DBDep, + admin: SudoAdminDep): + db_node = crud.get_node_by_id(db, node_id) + if not db_node: raise HTTPException(status_code=404, detail="Node not found") - return dbnode + return db_node -@app.websocket("/api/node/{node_id}/logs") -async def node_logs(node_id: int, websocket: WebSocket, db: Session = Depends(get_db)): +@router.websocket("/api/node/{node_id}/logs", dependencies=None) +async def node_logs(node_id: int, + websocket: WebSocket, + db: DBDep): token = ( websocket.query_params.get('token') or @@ -100,23 +95,17 @@ async def node_logs(node_id: int, websocket: WebSocket, db: Session = Depends(ge await websocket.send_text(l) -@app.get("/api/nodes", tags=['Node'], response_model=List[NodeResponse]) -def get_nodes(db: Session = Depends(get_db), - admin: Admin = Depends(Admin.get_current)): - if not admin.is_sudo: - raise HTTPException(status_code=403, detail="You're not allowed") +@router.get("/api/nodes", response_model=List[NodeResponse]) +def get_nodes(db: DBDep, + admin: SudoAdminDep): return crud.get_nodes(db) -@app.put("/api/node/{node_id}", tags=['Node'], response_model=NodeResponse) +@router.put("/api/node/{node_id}", response_model=NodeResponse) async def modify_node(node_id: int, modified_node: NodeModify, - db: Session = Depends(get_db), - admin: Admin = Depends(Admin.get_current)): - - if not admin.is_sudo: - raise HTTPException(status_code=403, detail="You're not allowed") - + db: DBDep, + admin: SudoAdminDep): db_node = crud.get_node_by_id(db, node_id) if not db_node: raise HTTPException(status_code=404, detail="Node not found") @@ -130,18 +119,14 @@ async def modify_node(node_id: int, ssl_key=certificate.key, ssl_cert=certificate.certificate) await marznode.operations.add_node(db_node.id, node) - logger.info(f"Node \"{db_node.name}\" modified") + logger.info("Node `%s` modified", db_node.name) return db_node -@app.post("/api/node/{node_id}/resync", tags=['Node']) +@router.post("/api/node/{node_id}/resync") async def reconnect_node(node_id: int, - db: Session = Depends(get_db), - admin: Admin = Depends(Admin.get_current)): - - if not admin.is_sudo: - raise HTTPException(status_code=403, detail="You're not allowed") - + db: DBDep, + admin: SudoAdminDep): db_node = crud.get_node_by_id(db, node_id) if not db_node: raise HTTPException(status_code=404, detail="Node not found") @@ -149,14 +134,10 @@ async def reconnect_node(node_id: int, return {} -@app.delete("/api/node/{node_id}", tags=['Node']) +@router.delete("/api/node/{node_id}") async def remove_node(node_id: int, - db: Session = Depends(get_db), - admin: Admin = Depends(Admin.get_current)): - - if not admin.is_sudo: - raise HTTPException(status_code=403, detail="You're not allowed") - + db: DBDep, + admin: SudoAdminDep): db_node = crud.get_node_by_id(db, node_id) if not db_node: raise HTTPException(status_code=404, detail="Node not found") @@ -164,32 +145,18 @@ async def remove_node(node_id: int, crud.remove_node(db, db_node) await marznode.operations.remove_node(db_node.id) - logger.info(f"Node \"{db_node.name}\" deleted") + logger.info(f"Node `%s` deleted", db_node.name) return {} -@app.get("/api/nodes/usage", tags=['Node'], response_model=NodesUsageResponse) -def get_usage(db: Session = Depends(get_db), - start: str = None, - end: str = None, - admin: Admin = Depends(Admin.get_current)): +@router.get("/api/nodes/usage", response_model=NodesUsageResponse) +def get_usage(db: DBDep, + admin: SudoAdminDep, + start_date: StartDateDep, + end_date: EndDateDep): """ Get nodes usage """ - - if not admin.is_sudo: - raise HTTPException(status_code=403, detail="You're not allowed") - - if start is None: - start_date = datetime.fromtimestamp(datetime.utcnow().timestamp() - 30 * 24 * 3600) - else: - start_date = datetime.fromisoformat(start) - - if end is None: - end_date = datetime.utcnow() - else: - end_date = datetime.fromisoformat(end) - usages = crud.get_nodes_usage(db, start_date, end_date) return {"usages": usages} diff --git a/app/routes/service.py b/app/routes/service.py new file mode 100644 index 0000000..752b219 --- /dev/null +++ b/app/routes/service.py @@ -0,0 +1,85 @@ +from typing import List + +import sqlalchemy +from fastapi import APIRouter, Depends +from fastapi import HTTPException + +from app.db import crud +from app.dependencies import DBDep, sudo_admin +from app.models.service import (ServiceCreate, ServiceModify, + ServiceResponse) + +router = APIRouter(dependencies=[Depends(sudo_admin)], tags=['Service']) + + +@router.post("/api/service", response_model=ServiceResponse) +def add_service(new_service: ServiceCreate, + db: DBDep): + """ + Add a new user template + + - **name** can be up to 64 characters + - **data_limit** must be in bytes and larger or equal to 0 + - **expire_duration** must be in seconds and larger or equal to 0 + - **inbounds** dictionary of protocol:inbound_tags, empty means all inbounds + """ + try: + return crud.create_service(db, new_service) + except sqlalchemy.exc.IntegrityError: + db.rollback() + raise HTTPException(status_code=409, detail="Service by this name already exists") + + +@router.get("/api/service/{id}", response_model=ServiceResponse) +def get_service(id: int, db: DBDep): + """ + Get Service information with id + """ + dbservice = crud.get_service(db, id) + if not dbservice: + raise HTTPException(status_code=404, detail="Service not found") + + return dbservice + + +@router.put("/api/service/{id}", response_model=ServiceResponse) +def modify_service(id: int, + modification: ServiceModify, + db: DBDep): + """ + Modify Service + + - **name** can be up to 64 characters + - **data_limit** must be in bytes and larger or equal to 0 + - **expire_duration** must be in seconds and larger or equat to 0 + - **inbounds** list of inbound ids. if not specified no change will be applied; + in case of an empty list all inbounds would be removed. + """ + # TODO: Update all affected users in nodes + dbservice = crud.get_service(db, id) + if not dbservice: + raise HTTPException(status_code=404, detail="Service not found") + + try: + return crud.update_service(db, dbservice, modification) + except sqlalchemy.exc.IntegrityError: + db.rollback() + raise HTTPException(status_code=409, detail="Service by this name already exists") + + +@router.delete("/api/service/{id}") +def remove_service(id: int, + db: DBDep): + dbservice = crud.get_service(db, id) + if not dbservice: + raise HTTPException(status_code=404, detail="Service not found") + + crud.remove_service(db, dbservice) + return dict() + + +@router.get("/api/services", response_model=List[ServiceResponse]) +def get_services(db: DBDep, + offset: int = None, + limit: int = None): + return crud.get_services(db) # , offset, limit) diff --git a/app/views/subscription.py b/app/routes/subscription.py similarity index 56% rename from app/views/subscription.py rename to app/routes/subscription.py index 6b1d5b4..fe5bfb0 100644 --- a/app/views/subscription.py +++ b/app/routes/subscription.py @@ -1,58 +1,56 @@ import re -from datetime import datetime -from fastapi import Depends, Header, HTTPException, Path, Request, Response +from fastapi import APIRouter +from fastapi import Header, HTTPException, Path, Request, Response from fastapi.responses import HTMLResponse -from app import app -from app.db import Session, crud, get_db +from app.db import crud +from app.dependencies import DBDep, SubUserDep, StartDateDep, EndDateDep from app.models.user import UserResponse from app.templates import render_template -from app.utils.share import encode_title, generate_subscription +from app.utils.share import encode_title, generate_subscription, generate_v2ray_links from config import ( SUB_PROFILE_TITLE, SUB_SUPPORT_URL, SUB_UPDATE_INTERVAL, - SUBSCRIPTION_PAGE_TEMPLATE, - XRAY_SUBSCRIPTION_PATH + SUBSCRIPTION_PAGE_TEMPLATE ) +router = APIRouter(tags=['Subscription']) -@app.get("/%s/{username}/{key}/" % XRAY_SUBSCRIPTION_PATH, tags=['Subscription']) -@app.get("/%s/{username}/{key}" % XRAY_SUBSCRIPTION_PATH, include_in_schema=False) -def user_subscription(username: str, - key: str, + +def get_subscription_user_info(user: UserResponse) -> dict: + return { + "upload": 0, + "download": user.used_traffic, + "total": user.data_limit, + "expire": user.expire, + } + + +@router.get("/sub/{username}/{key}/") +@router.get("/sub/{username}/{key}", include_in_schema=False) +def user_subscription(db_user: SubUserDep, request: Request, - db: Session = Depends(get_db), + db: DBDep, user_agent: str = Header(default="")): """ Subscription link, V2ray and Clash supported """ accept_header = request.headers.get("Accept", "") - def get_subscription_user_info(user: UserResponse) -> dict: - return { - "upload": 0, - "download": user.used_traffic, - "total": user.data_limit, - "expire": user.expire, - } - try: - int(key, 16) - except ValueError: - return Response(status_code=404) - - dbuser = crud.get_user(db, username) - if not dbuser or dbuser.key != key: - return Response(status_code=404) - - user: UserResponse = UserResponse.model_validate(dbuser) + user: UserResponse = UserResponse.model_validate(db_user) if "text/html" in accept_header: + links = generate_v2ray_links(user.inbounds, + user.key, + user.model_dump( + exclude={"subscription_url", "services", "inbounds", "links"} + )) return HTMLResponse( render_template( SUBSCRIPTION_PAGE_TEMPLATE, - {"user": user} + {"user": user, "links": links} ) ) @@ -69,7 +67,7 @@ def user_subscription(username: str, ) } - crud.update_user_sub(db, dbuser, user_agent) + crud.update_user_sub(db, db_user, user_agent) if re.match('^([Cc]lash-verge|[Cc]lash-?[Mm]eta)', user_agent): conf = generate_subscription(user=user, config_format="clash-meta", as_base64=False) @@ -92,83 +90,31 @@ def user_subscription(username: str, return Response(content=conf, media_type="text/plain", headers=response_headers) -@app.get("/%s/{username}/{key}/info" % XRAY_SUBSCRIPTION_PATH, tags=['Subscription'], response_model=UserResponse) -def user_subscription_info(username: str, - key: str, - db: Session = Depends(get_db)): - try: - int(key, 16) - except ValueError: - return Response(status_code=404) - - dbuser = crud.get_user(db, username) - if not dbuser or key != dbuser.key: - return Response(status_code=404) - - return dbuser - +@router.get("/sub/{username}/{key}/info", response_model=UserResponse) +def user_subscription_info(db_user: SubUserDep): + return db_user -@app.get("/%s/{username}/{key}/usage" % XRAY_SUBSCRIPTION_PATH, tags=['Subscription']) -def user_get_usage(username: str, - key: str, - start: str = None, - end: str = None, - db: Session = Depends(get_db)): - try: - int(key, 16) - except ValueError: - return Response(status_code=404) +@router.get("/sub/{username}/{key}/usage") +def user_get_usage(db_user: SubUserDep, + db: DBDep, + start_date: StartDateDep, + end_date: EndDateDep): + usages = crud.get_user_usages(db, db_user, start_date, end_date) - dbuser = crud.get_user(db, username) - if not dbuser or key != dbuser.key: - return Response(status_code=404) + return {"usages": usages, "username": db_user.username} - if start is None: - start_date = datetime.fromtimestamp(datetime.utcnow().timestamp() - 30 * 24 * 3600) - else: - start_date = datetime.fromisoformat(start) - - if end is None: - end_date = datetime.utcnow() - else: - end_date = datetime.fromisoformat(end) - - usages = crud.get_user_usages(db, dbuser, start_date, end_date) - - return {"usages": usages, "username": dbuser.username} - -@app.get("/%s/{username}/{key}/{client_type}" % XRAY_SUBSCRIPTION_PATH, tags=['Subscription']) +@router.get("/sub/{username}/{key}/{client_type}") def user_subscription_with_client_type( - username: str, - key: str, + db_user: SubUserDep, request: Request, - client_type: str = Path(..., regex="sing-box|clash-meta|clash|outline|v2ray"), - db: Session = Depends(get_db), -): + client_type: str = Path(..., regex="sing-box|clash-meta|clash|outline|v2ray")): """ Subscription link, v2ray, clash, sing-box, outline and clash-meta supported """ - def get_subscription_user_info(user: UserResponse) -> dict: - return { - "upload": 0, - "download": user.used_traffic, - "total": user.data_limit, - "expire": user.expire, - } - - try: - int(key, 16) - except ValueError: - return Response(status_code=404) - - dbuser = crud.get_user(db, username) - if not dbuser or key != dbuser.key: - return Response(status_code=404) - - user: UserResponse = UserResponse.model_validate(dbuser) + user: UserResponse = UserResponse.model_validate(db_user) response_headers = { "content-disposition": f'attachment; filename="{user.username}"', diff --git a/app/routes/system.py b/app/routes/system.py new file mode 100644 index 0000000..0666a15 --- /dev/null +++ b/app/routes/system.py @@ -0,0 +1,75 @@ +from typing import Dict, List, Union + +from fastapi import APIRouter + +from app import __version__ +# from .. import __version__ +from app.db import crud +from app.dependencies import DBDep, AdminDep, SudoAdminDep +from app.models.admin import Admin +from app.models.proxy import InboundHost, Inbound +from app.models.system import SystemStats +from app.models.user import UserStatus +from app.utils.system import memory_usage, cpu_usage, realtime_bandwidth + +router = APIRouter(tags=["System"]) + + +@router.get("/api/system", response_model=SystemStats) +def get_system_stats(db: DBDep, admin: AdminDep): + mem = memory_usage() + cpu = cpu_usage() + system = crud.get_system_usage(db) + dbadmin: Union[Admin, None] = crud.get_admin(db, admin.username) + + total_user = crud.get_users_count(db, admin=dbadmin if not admin.is_sudo else None) + users_active = crud.get_users_count(db, status=UserStatus.active, admin=dbadmin if not admin.is_sudo else None) + realtime_bandwidth_stats = realtime_bandwidth() + + return SystemStats( + version=__version__, + mem_total=mem.total, + mem_used=mem.used, + cpu_cores=cpu.cores, + cpu_usage=cpu.percent, + total_user=total_user, + users_active=users_active, + incoming_bandwidth=system.uplink, + outgoing_bandwidth=system.downlink, + incoming_bandwidth_speed=realtime_bandwidth_stats.incoming_bytes, + outgoing_bandwidth_speed=realtime_bandwidth_stats.outgoing_bytes, + ) + + +@router.get('/api/inbounds', response_model=List[Inbound]) +def get_inbounds(db: DBDep, admin: SudoAdminDep): + inbounds = crud.get_all_inbounds(db) + return inbounds + + +@router.get('/api/hosts', response_model=Dict[int, List[InboundHost]]) +def get_hosts(db: DBDep, admin: SudoAdminDep): + hosts = crud.get_all_hosts(db) + + return hosts + + +@router.put('/api/hosts', response_model=Dict[str, List[InboundHost]]) +def modify_hosts(modified_hosts: Dict[str, List[InboundHost]], + db: DBDep, + admin: SudoAdminDep): + # validate + # for inbound_id, hosts in modified_hosts.items(): + # if not xray.config.inbounds_by_id.get(inbound_id): + # raise HTTPException(status_code=400, detail=f"Inbound {inbound_id} doesn't exist") + + for inbound_id, hosts in modified_hosts.items(): + crud.update_hosts(db, inbound_id, hosts) + + # xray.hosts.update() + + hosts = {} + # for inbound_tag in xray.config.inbounds_by_tag: + # hosts[inbound_tag] = crud.get_hosts(db, inbound_tag) + + return hosts diff --git a/app/routes/user.py b/app/routes/user.py new file mode 100644 index 0000000..49f9a13 --- /dev/null +++ b/app/routes/user.py @@ -0,0 +1,302 @@ +import asyncio +import logging +from datetime import datetime, timezone +from typing import List + +import sqlalchemy +from fastapi import APIRouter +from fastapi import HTTPException, Query + +from app import marznode +from app.db import crud +from app.dependencies import DBDep, AdminDep, SudoAdminDep, UserDep, StartDateDep, EndDateDep +from app.models.user import (UserCreate, UserModify, UserResponse, + UsersResponse, UserStatus, UserUsagesResponse) +from app.utils import report + +logger = logging.getLogger(__name__) +router = APIRouter(tags=['User']) + + +@router.post("/api/user", response_model=UserResponse) +async def add_user(new_user: UserCreate, + db: DBDep, + admin: AdminDep): + """ + Add a new user + + - **username** must have 3 to 32 characters and is allowed to contain a-z, 0-9, and underscores in between + - **expire** must be a UTC timestamp + - **data_limit** must be in Bytes, e.g. 1073741824B = 1GB + - **services** list of service ids + """ + # TODO expire should be datetime instead of timestamp + + try: + db_user = crud.create_user(db, new_user, + admin=crud.get_admin(db, admin.username)) + except sqlalchemy.exc.IntegrityError: + db.rollback() + raise HTTPException(status_code=409, detail="User already exists") + + user = UserResponse.model_validate(db_user) + await marznode.operations.add_user(user=db_user) + asyncio.create_task( + report.user_created( + user=user, + user_id=db_user.id, + by=admin + )) + logger.info("New user `%s` added", db_user.username) + return user + + +@router.get("/api/user/{username}", response_model=UserResponse) +def get_user(db_user: UserDep): + """ + Get users information + """ + return db_user + + +@router.put("/api/user/{username}", response_model=UserResponse) +async def modify_user(db_user: UserDep, + modifications: UserModify, + db: DBDep, + admin: AdminDep): + """ + Modify a user + + - set **expire** to 0 to make the user unlimited in time, null for no change + - set **data_limit** to 0 to make the user unlimited in data, null for no change + """ + old_status = db_user.status + status_change = bool(modifications.status and modifications.status != db_user.status) + if status_change and (modifications.status == UserStatus.disabled and + old_status in {UserStatus.active, UserStatus.on_hold}): + await marznode.operations.remove_user(db_user) + + old_inbounds = {(i.node_id, i.protocol, i.tag) for i in db_user.inbounds} + new_user = crud.update_user(db, db_user, modifications) + new_inbounds = {(i.node_id, i.protocol, i.tag) for i in new_user.inbounds} + user = UserResponse.model_validate(db_user) + + inbound_change = old_inbounds != new_inbounds + if (user.status in {UserStatus.active, UserStatus.on_hold} + and (not status_change or old_status == UserStatus.on_hold)): + if inbound_change: + await marznode.operations.update_user_inbounds( + user=new_user, new_inbounds=new_inbounds, old_inbounds=old_inbounds) + elif (user.status in {UserStatus.active, UserStatus.on_hold} and status_change and + old_status not in {UserStatus.active, UserStatus.on_hold}): + await marznode.operations.add_user(new_user) + + asyncio.create_task(report.user_updated( + user=user, + by=admin)) + + logger.info("User `%s` modified", user.username) + + if status_change: + asyncio.create_task(report.status_change( + username=user.username, + status=user.status, + user=user, + by=admin)) + + logger.info("User `%s` status changed from `%s` to `%s`", + user.username, + old_status, + user.status) + + return user + + +@router.delete("/api/user/{username}") +async def remove_user(db_user: UserDep, + db: DBDep, + admin: AdminDep): + """ + Remove a user + """ + await marznode.operations.remove_user(db_user) + + crud.remove_user(db, db_user) + db.flush() + + asyncio.create_task( + report.user_deleted( + username=db_user.username, + by=admin + )) + logger.info("User %s deleted", db_user.username) + return {} + + +@router.post("/api/user/{username}/reset", response_model=UserResponse) +async def reset_user_data_usage(db_user: UserDep, + db: DBDep, + admin: AdminDep): + """ + Reset user data usage + """ + previous_status = db_user.status + db_user = crud.reset_user_data_usage(db, db_user) + + if db_user.status == UserStatus.active and previous_status == UserStatus.limited: + await marznode.operations.add_user(db_user) + + user = UserResponse.model_validate(db_user) + asyncio.create_task(report.user_data_usage_reset( + user=user, + by=admin)) + + logger.info("User `%s`'s usage was reset", db_user.username) + + return user + + +@router.post("/api/user/{username}/revoke_sub", response_model=UserResponse) +async def revoke_user_subscription(db_user: UserDep, + db: DBDep, + admin: AdminDep): + """ + Revoke users subscription (Subscription link and proxies) + """ + db_user = crud.revoke_user_sub(db, db_user) + + if db_user.status in [UserStatus.active, UserStatus.on_hold]: + await marznode.operations.remove_user(db_user) + await marznode.operations.add_user(db_user) + user = UserResponse.model_validate(db_user) + asyncio.create_task( + report.user_subscription_revoked( + user=user, + by=admin + )) + + logger.info("User %s subscription revoked", db_user.username) + + return user + + +@router.get("/api/users", response_model=UsersResponse) +def get_users(db: DBDep, + admin: AdminDep, + offset: int = None, + limit: int = None, + username: List[str] = Query(None), + status: UserStatus = None, + sort: str = None): + """ + Get all users + """ + dbadmin = crud.get_admin(db, admin.username) + + if sort is not None: + opts = sort.strip(',').split(',') + sort = [] + for opt in opts: + try: + sort.append(crud.UsersSortingOptions[opt]) + except KeyError: + raise HTTPException(status_code=400, + detail=f'"{opt}" is not a valid sort option') + + users, count = crud.get_users(db=db, + offset=offset, + limit=limit, + usernames=username, + status=status, + sort=sort, + admin=dbadmin if not admin.is_sudo else None, + return_with_count=True) + + return {"users": users, "total": count} + + +@router.post("/api/users/reset") +async def reset_users_data_usage(db: DBDep, + admin: SudoAdminDep): + """ + Reset all users data usage + """ + dbadmin = crud.get_admin(db, admin.username) + crud.reset_all_users_data_usage(db=db, admin=dbadmin) + # startup_config = xray.config.include_db_users() + await xray.core.restart(xray.config.include_db_users(node_id=0)) + for node_id, node in list(xray.nodes.items()): + if node.connected: + xray.operations.restart_node(node_id, xray.config.include_db_users(node_id=node_id)) + return {} + + +@router.get("/api/user/{username}/usage", response_model=UserUsagesResponse) +def get_user_usage(db: DBDep, + db_user: UserDep, + start_date: StartDateDep, + end_date: EndDateDep): + """ + Get users usage + """ + usages = crud.get_user_usages(db, db_user, start_date, end_date) + + return {"usages": usages, "username": db_user.username} + + +@router.put("/api/user/{username}/set-owner", response_model=UserResponse) +def set_owner(username: str, + admin_username: str, + db: DBDep, + admin: SudoAdminDep): + db_user = crud.get_user(db, username) + if not db_user: + raise HTTPException(status_code=404, detail="User not found") + + new_admin = crud.get_admin(db, username=admin_username) + if not new_admin: + raise HTTPException(status_code=404, detail="Admin not found") + + db_user = crud.set_owner(db, db_user, new_admin) + user = UserResponse.model_validate(db_user) + + logger.info("`%s`'s owner successfully set to `%s`", user.username, admin.username) + + return user + + +@router.delete("/api/users/expired") +def delete_expired(passed_time: int, + db: DBDep, + admin: AdminDep): + """ + Delete expired users + - **passed_time** must be a timestamp + - This function will delete all expired users that meet the specified number of days passed and can't be undone. + """ + + dbadmin = crud.get_admin(db, admin.username) + + db_users = crud.get_users(db=db, + status=[UserStatus.expired, UserStatus.limited], + admin=dbadmin if not admin.is_sudo else None) + + current_time = int(datetime.now(timezone.utc).timestamp()) + expiration_threshold = current_time - passed_time + expired_users = [ + user for user in db_users if user.expire is not None and user.expire <= expiration_threshold] + if not expired_users: + raise HTTPException(status_code=404, detail="No expired user found.") + + for db_user in expired_users: + crud.remove_user(db, db_user) + + asyncio.get_running_loop().create_task( + report.user_deleted( + username=db_user.username, + by=admin + )) + + logger.info("User `%s` removed", db_user.username) + + return {} diff --git a/app/tasks/__init__.py b/app/tasks/__init__.py new file mode 100644 index 0000000..1396639 --- /dev/null +++ b/app/tasks/__init__.py @@ -0,0 +1,19 @@ +import logging +from .nodes import nodes_startup, nodes_health_check +from .record_usages import record_user_usages +from .reset_user_data_usage import reset_user_data_usage +from .review_users import review_users +from .send_notifications import delete_expired_reminders, send_notifications +from .store_bandwidth import record_realtime_bandwidth + +__all__ = [ + "nodes_startup", + "nodes_health_check", + "record_user_usages", + "reset_user_data_usage", + "review_users", + "delete_expired_reminders", + "send_notifications", + "store_bandwidth", + "record_realtime_bandwidth" +] diff --git a/app/jobs/1_start_nodes.py b/app/tasks/nodes.py similarity index 92% rename from app/jobs/1_start_nodes.py rename to app/tasks/nodes.py index a9dcc2b..a16b64d 100644 --- a/app/jobs/1_start_nodes.py +++ b/app/tasks/nodes.py @@ -2,7 +2,7 @@ from collections import defaultdict from grpclib import GRPCError -from app import app, scheduler, marznode +from app import marznode from app.db import GetDB, crud, get_tls_certificate from app.marznode import MarzNodeGRPC from app.models.node import NodeStatus @@ -35,7 +35,6 @@ async def sync_node(db, node, node_id): node.synced = True -@app.on_event("startup") async def nodes_startup(): with GetDB() as db: certificate = get_tls_certificate(db) @@ -51,4 +50,3 @@ async def nodes_startup(): await marznode.operations.add_node(db_node.id, node) # crud.update_node_status(db, db_node, NodeStatus.connecting) - scheduler.add_job(nodes_health_check, 'interval', seconds=5, coalesce=True, max_instances=1) diff --git a/app/jobs/record_usages.py b/app/tasks/record_usages.py similarity index 97% rename from app/jobs/record_usages.py rename to app/tasks/record_usages.py index 7420493..d15d67d 100644 --- a/app/jobs/record_usages.py +++ b/app/tasks/record_usages.py @@ -5,7 +5,7 @@ from datetime import datetime from grpclib import GRPCError from sqlalchemy import and_, select, insert, update, bindparam -from app import marznode, scheduler +from app import marznode from app.db import GetDB from app.db.models import NodeUsage, NodeUserUsage, User from app.marznode import MarzNodeBase @@ -119,4 +119,3 @@ async def record_user_usages(): record_user_stats(params, node_id, node.usage_coefficient if (node := marznode.nodes.get(node_id)) else 1) -scheduler.add_job(record_user_usages, 'interval', coalesce=True, seconds=30) diff --git a/app/jobs/reset_user_data_usage.py b/app/tasks/reset_user_data_usage.py similarity index 91% rename from app/jobs/reset_user_data_usage.py rename to app/tasks/reset_user_data_usage.py index 3ce2c25..196d66c 100644 --- a/app/jobs/reset_user_data_usage.py +++ b/app/tasks/reset_user_data_usage.py @@ -1,9 +1,12 @@ +import logging from datetime import datetime -from app import logger, scheduler, marznode +from app import marznode from app.db import crud, GetDB, get_users from app.models.user import UserDataLimitResetStrategy, UserStatus +logger = logging.getLogger(__name__) + reset_strategy_to_days = { UserDataLimitResetStrategy.day.value: 1, UserDataLimitResetStrategy.week.value: 7, @@ -34,6 +37,3 @@ async def reset_user_data_usage(): await marznode.operations.add_user(user) logger.info(f"User data usage reset for User \"{user.username}\"") - - -scheduler.add_job(reset_user_data_usage, 'interval', coalesce=True, hours=1) diff --git a/app/jobs/review_users.py b/app/tasks/review_users.py similarity index 95% rename from app/jobs/review_users.py rename to app/tasks/review_users.py index 2748997..b02fd92 100644 --- a/app/jobs/review_users.py +++ b/app/tasks/review_users.py @@ -1,10 +1,11 @@ import asyncio +import logging from datetime import datetime from typing import TYPE_CHECKING from sqlalchemy.orm import Session -from app import logger, scheduler, marznode +from app import marznode from app.db import (GetDB, get_notification_reminder, get_users, start_user_expire, update_user_status) from app.models.user import ReminderType, UserResponse, UserStatus @@ -18,6 +19,9 @@ if TYPE_CHECKING: from app.db.models import User +logger = logging.getLogger(__name__) + + def add_notification_reminders(db: Session, user: "User", now: datetime = datetime.utcnow()) -> None: if user.data_limit: usage_percent = calculate_usage_percent( @@ -35,7 +39,7 @@ def add_notification_reminders(db: Session, user: "User", now: datetime = dateti user.id, user.expire) -async def review(): +async def review_users(): now = datetime.utcnow() with GetDB() as db: for user in get_users(db, status=UserStatus.active): @@ -88,4 +92,3 @@ async def review(): logger.info(f"on hold user `{user.username}` has been activated") -scheduler.add_job(review, 'interval', seconds=600, coalesce=True, max_instances=1) diff --git a/app/jobs/send_notifications.py b/app/tasks/send_notifications.py similarity index 83% rename from app/jobs/send_notifications.py rename to app/tasks/send_notifications.py index 2a279b2..f1a75e0 100644 --- a/app/jobs/send_notifications.py +++ b/app/tasks/send_notifications.py @@ -1,15 +1,18 @@ +import logging from datetime import datetime as dt from datetime import timedelta as td from typing import Any, Dict, List +import aiohttp from fastapi.encoders import jsonable_encoder from requests import Session import config -from app import app, logger, scheduler -from app.db import GetDB from app.db.models import NotificationReminder from app.utils.notification import queue +from app.db import GetDB + +logger = logging.getLogger(__name__) session = Session() @@ -70,13 +73,3 @@ def delete_expired_reminders() -> None: db.query(NotificationReminder).filter(NotificationReminder.expires_at < dt.utcnow()).delete() db.commit() - -if config.WEBHOOK_ADDRESS: - @app.on_event("shutdown") - async def app_shutdown(): - logger.info("Sending pending notifications before shutdown...") - await send_notifications() - - logger.info("Send webhook job started") - scheduler.add_job(send_notifications, "interval", seconds=30, replace_existing=True) - scheduler.add_job(delete_expired_reminders, "interval", hours=2, start_date=dt.utcnow() + td(minutes=1)) diff --git a/app/tasks/store_bandwidth.py b/app/tasks/store_bandwidth.py new file mode 100644 index 0000000..9152b35 --- /dev/null +++ b/app/tasks/store_bandwidth.py @@ -0,0 +1,11 @@ +import psutil + +from app.utils.system import rt_bw + + +def record_realtime_bandwidth() -> None: + io = psutil.net_io_counters() + rt_bw.incoming_bytes, rt_bw.bytes_recv = io.bytes_recv - rt_bw.bytes_recv, io.bytes_recv + rt_bw.outgoing_bytes, rt_bw.bytes_sent = io.bytes_sent - rt_bw.bytes_sent, io.bytes_sent + rt_bw.incoming_packets, rt_bw.packets_recv = io.packets_recv - rt_bw.packets_recv, io.packets_recv + rt_bw.outgoing_packets, rt_bw.packet_sent = io.packets_sent - rt_bw.packet_sent, io.packets_sent \ No newline at end of file diff --git a/app/telegram/__init__.py b/app/telegram/__init__.py index 6b087a1..f95e6d3 100644 --- a/app/telegram/__init__.py +++ b/app/telegram/__init__.py @@ -1,23 +1,17 @@ import importlib.util -import asyncio from os.path import dirname from config import TELEGRAM_API_TOKEN, TELEGRAM_PROXY_URL -from app import app from telebot.async_telebot import AsyncTeleBot -from telebot import asyncio_helper #, logger - -# import logging - -# logger.setLevel(logging.DEBUG) +from telebot import asyncio_helper bot = None if TELEGRAM_API_TOKEN: asyncio_helper.proxy = TELEGRAM_PROXY_URL bot = AsyncTeleBot(TELEGRAM_API_TOKEN) -handler_names = ["admin", "report", "user"] +handler_names = ["admin", "user"] + -@app.on_event("startup") async def start_bot(): if bot: handler_dir = dirname(__file__) + "/handlers/" @@ -27,7 +21,7 @@ async def start_bot(): from app.telegram import utils # setup custom handlers utils.setup() - asyncio.get_running_loop().create_task(bot.polling()) + await bot.polling() from .handlers.report import ( # noqa diff --git a/app/telegram/handlers/report.py b/app/telegram/handlers/report.py index 4cc977d..951ad32 100644 --- a/app/telegram/handlers/report.py +++ b/app/telegram/handlers/report.py @@ -1,13 +1,16 @@ import datetime +import logging +from datetime import datetime -from app import logger -from app.telegram import bot from telebot.apihelper import ApiTelegramException -from datetime import datetime +from telebot.formatting import escape_html + +from app.telegram import bot from app.telegram.utils.keyboard import BotKeyboard from app.utils.system import readable_size from config import TELEGRAM_ADMIN_ID, TELEGRAM_LOGGER_CHANNEL_ID -from telebot.formatting import escape_html + +logger = logging.getLogger(__name__) async def report(message: str, parse_mode="html", keyboard=None): diff --git a/app/telegram/utils/__init__.py b/app/telegram/utils/__init__.py index be27122..6699e5e 100644 --- a/app/telegram/utils/__init__.py +++ b/app/telegram/utils/__init__.py @@ -2,4 +2,4 @@ from app.telegram.utils import custom_filters def setup() -> None: - custom_filters.setup() \ No newline at end of file + custom_filters.setup() diff --git a/app/telegram/utils/custom_filters.py b/app/telegram/utils/custom_filters.py index e5db305..c8965ae 100644 --- a/app/telegram/utils/custom_filters.py +++ b/app/telegram/utils/custom_filters.py @@ -1,7 +1,7 @@ from app.telegram import bot from typing import Union -from telebot import types, custom_filters +from telebot import types from telebot.asyncio_filters import AdvancedCustomFilter, StateFilter from config import TELEGRAM_ADMIN_ID @@ -30,7 +30,6 @@ def query_startswith(text: Union[list, str]): return lambda query: query.data.startswith(tuple(text)) - def setup() -> None: bot.add_custom_filter(IsAdminFilter()) bot.add_custom_filter(StateFilter(bot)) diff --git a/app/templates/__init__.py b/app/templates/__init__.py index 8d0c2ba..4916429 100644 --- a/app/templates/__init__.py +++ b/app/templates/__init__.py @@ -4,7 +4,6 @@ from typing import Union import jinja2 from config import CUSTOM_TEMPLATES_DIRECTORY - from .filters import CUSTOM_FILTERS template_directories = ["app/templates"] diff --git a/app/templates/subscription/index.html b/app/templates/subscription/index.html index 98a442e..f26f825 100644 --- a/app/templates/subscription/index.html +++ b/app/templates/subscription/index.html @@ -94,7 +94,7 @@ {% if user.status == 'active' %}

Links: