diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
rename from app/views/admin.py
rename to 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
rename from app/views/node.py
rename to 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
@@ -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
rename from app/views/subscription.py
rename to 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
@@ -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
@@ -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
@@ -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
rename from app/jobs/1_start_nodes.py
rename to 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
rename from app/jobs/record_usages.py
rename to 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
rename from app/jobs/reset_user_data_usage.py
rename to 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
rename from app/jobs/review_users.py
rename to 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
rename from app/jobs/send_notifications.py
rename to 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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -94,7 +94,7 @@
{% if user.status == 'active' %}
<h2>Links:</h2>
<ul>
- {% for link in user.links %}
+ {% for link in links %}
<li>
<input type="text" value="{{ link }}" readonly>
<button class="copy-button" onclick="copyLink('{{ link }}', this)">Copy</button>
diff --git a/app/utils/jwt.py b/app/utils/jwt.py
@@ -4,12 +4,12 @@ from typing import Union
from jose import JWTError, jwt
-from config import JWT_ACCESS_TOKEN_EXPIRE_MINUTES
+from config import JWT_ACCESS_TOKEN_EXPIRE_MINUTES, SUDOERS
+from app.db import GetDB, get_jwt_secret_key
@lru_cache(maxsize=None)
def get_secret_key():
- from app.db import GetDB, get_jwt_secret_key
with GetDB() as db:
return get_jwt_secret_key(db)
diff --git a/app/utils/system.py b/app/utils/system.py
@@ -1,4 +1,3 @@
-import requests
import math
import secrets
import socket
@@ -6,18 +5,16 @@ from dataclasses import dataclass
import psutil
-from app import scheduler
-
@dataclass
-class MemoryStat():
+class MemoryStat:
total: int
used: int
free: int
@dataclass
-class CPUStat():
+class CPUStat:
cores: int
percent: float
@@ -63,15 +60,6 @@ rt_bw = RealtimeBandwidth(
incoming_bytes=0, outgoing_bytes=0, incoming_packets=0, outgoing_packets=0)
-@scheduler.scheduled_job('interval', seconds=1, coalesce=True, max_instances=1)
-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
-
-
def realtime_bandwidth() -> RealtimeBandwidthStat:
return RealtimeBandwidthStat(
incoming_bytes=rt_bw.incoming_bytes, outgoing_bytes=rt_bw.outgoing_bytes,
@@ -94,24 +82,6 @@ def check_port(port: int) -> bool:
def get_public_ip():
- """
- try:
- return requests.get('https://api.ipify.org?format=json&ipv=4', timeout=5).json()['ip']
- except (requests.exceptions.RequestException,
- requests.exceptions.RequestException,
- KeyError) as e:
- pass
-
- try:
- requests.packages.urllib3.util.connection.HAS_IPV6 = False
- return requests.get('https://ifconfig.io/ip', timeout=5).text.strip()
- except (requests.exceptions.RequestException,
- requests.exceptions.RequestException,
- KeyError) as e:
- pass
- finally:
- requests.packages.urllib3.util.connection.HAS_IPV6 = True
- """
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.connect(('8.8.8.8', 80))
diff --git a/app/views/__init__.py b/app/views/__init__.py
@@ -1,7 +0,0 @@
-from .admin import *
-from .subscription import *
-from .system import *
-from .user import *
-from .service import *
-from .node import *
-from .home import *
diff --git a/app/views/home.py b/app/views/home.py
@@ -1,10 +0,0 @@
-from fastapi.responses import HTMLResponse
-
-from app import app
-from app.templates import render_template
-from config import HOME_PAGE_TEMPLATE
-
-
-@app.get("/", response_class=HTMLResponse)
-def base():
- return render_template(HOME_PAGE_TEMPLATE)
diff --git a/app/views/service.py b/app/views/service.py
@@ -1,90 +0,0 @@
-from typing import List
-
-import sqlalchemy
-from fastapi import Depends, HTTPException
-
-from app import app
-from app.db import Session, crud, get_db
-from app.models.admin import Admin
-from app.models.service import (ServiceCreate, ServiceModify,
- ServiceResponse)
-
-
-@app.post("/api/service", tags=['Service'], response_model=ServiceResponse)
-def add_service(new_service: ServiceCreate,
- db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- 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 equat to 0
- - **inbounds** dictionary of protocol:inbound_tags, empty means all inbounds
- """
- if not (admin.is_sudo):
- raise HTTPException(status_code=403, detail="You're not allowed")
- 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")
-
-
-@app.get("/api/service/{id}", tags=['Service'], response_model=ServiceResponse)
-def get_service(id: int, db: Session = Depends(get_db), admin: Admin = Depends(Admin.get_current)):
- """
- 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
-
-
-@app.put("/api/service/{id}", tags=['Service'], response_model=ServiceResponse)
-def modify_service(id: int, modify_service: ServiceModify,
- db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- 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** dictionary of protocol:inbound_tags, empty means all inbounds
- """
- if not (admin.is_sudo):
- raise HTTPException(status_code=403, detail="You're not allowed")
-
- dbservice = crud.get_service(db, id)
- if not dbuser_template:
- raise HTTPException(status_code=404, detail="Service not found")
-
- try:
- return crud.update_service(db, dbservice, modify_service)
- except sqlalchemy.exc.IntegrityError:
- db.rollback()
- raise HTTPException(status_code=409, detail="Service by this name already exists")
-
-
-@app.delete("/api/service/{id}", tags=['Service'])
-def remove_service(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")
-
- dbservice = crud.get_service(db, id)
- if not dbuser_template:
- raise HTTPException(status_code=404, detail="Service not found")
-
- return crud.remove_service(db, dbservice)
-
-
-@app.get("/api/services", tags=['Service'], response_model=List[ServiceResponse])
-def get_services(
- offset: int = None, limit: int = None, db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- return crud.get_services(db) #, offset, limit)
diff --git a/app/views/system.py b/app/views/system.py
@@ -1,81 +0,0 @@
-from typing import Dict, List, Union
-
-from fastapi import Depends, HTTPException
-
-from app import app
-from app.db import Session, crud, get_db
-from app.models.admin import Admin
-from app.models.proxy import InboundHost, Inbound, ProxyTypes, InboundBase
-from app.models.system import SystemStats
-from app.models.user import UserStatus
-from app.utils.system import memory_usage, cpu_usage, realtime_bandwidth
-from app import __version__
-
-
-@app.get("/api/system", tags=["System"], response_model=SystemStats)
-def get_system_stats(db: Session = Depends(get_db), admin: Admin = Depends(Admin.get_current)):
- 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,
- )
-
-
-@app.get('/api/inbounds', tags=["System"], response_model=List[Inbound])
-def get_inbounds(db: Session = Depends(get_db), admin: Admin = Depends(Admin.get_current)):
- inbounds = crud.get_all_inbounds(db)# xray.config.inbounds_by_protocol
- return inbounds
-
-
-@app.get('/api/hosts', tags=["System"], response_model=Dict[int, List[InboundHost]])
-def get_hosts(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")
-
- # hosts = {}
- hosts = crud.get_all_hosts(db)
- #for inbound_tag in xray.config.inbounds_by_tag:
- # hosts[inbound_tag] = crud.get_hosts(db, inbound_tag)
-
- return hosts
-
-
-@app.put('/api/hosts', tags=["System"], response_model=Dict[str, List[InboundHost]])
-def modify_hosts(modified_hosts: Dict[str, List[InboundHost]],
- 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")
-
- # 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/views/user.py b/app/views/user.py
@@ -1,349 +0,0 @@
-from datetime import datetime, timezone
-from typing import List
-
-import asyncio
-import sqlalchemy
-from fastapi import BackgroundTasks, Depends, HTTPException, Query
-
-from app import app, logger, marznode
-from app.db import Session, crud, get_db
-from app.models.admin import Admin
-from app.models.user import (UserCreate, UserModify, UserResponse,
- UsersResponse, UserStatus, UserUsagesResponse)
-from app.utils import report
-
-
-@app.post("/api/user", tags=['User'], response_model=UserResponse)
-async def add_user(new_user: UserCreate,
- db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- 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 an UTC timestamp
- - **data_limit** must be in Bytes, e.g. 1073741824B = 1GB
- - **proxies** dictionary of protocol:settings
- - **inbounds** dictionary of protocol:inbound_tags, empty means all inbounds
- """
- # TODO expire should be datetime instead of timestamp
-
- try:
- dbuser = 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(dbuser)
- await marznode.operations.add_user(user=dbuser)
- asyncio.create_task(
- report.user_created(
- user=user,
- user_id=dbuser.id,
- by=admin
- ))
- logger.info(f"New user \"{dbuser.username}\" added")
- return user
-
-
-@app.get("/api/user/{username}", tags=['User'], response_model=UserResponse)
-def get_user(username: str,
- db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- Get users information
- """
- if not (admin.is_sudo or (dbuser.admin and dbuser.admin.username == admin.username)):
- raise HTTPException(status_code=403, detail="You're not allowed")
-
- dbuser = crud.get_user(db, username)
- if not dbuser:
- raise HTTPException(status_code=404, detail="User not found")
-
- return dbuser
-
-
-@app.put("/api/user/{username}", tags=['User'], response_model=UserResponse)
-async def modify_user(username: str,
- modified_user: UserModify,
- db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- Modify a user
-
- - set **expire** to 0 to make the user unlimited in time, null to no change
- - set **data_limit** to 0 to make the user unlimited in data, null to no change
- - **proxies** dictionary of protocol:settings, empty means no change
- - **inbounds** dictionary of protocol:inbound_tags, empty means no change
- """
- dbuser = crud.get_user(db, username)
- if not (admin.is_sudo or (dbuser.admin and dbuser.admin.username == admin.username)):
- raise HTTPException(status_code=403, detail="You're not allowed")
-
- if not dbuser:
- raise HTTPException(status_code=404, detail="User not found")
- old_inbounds = {(i.node_id, i.protocol, i.tag) for i in dbuser.inbounds}
- old_status = dbuser.status
- new_user = crud.update_user(db, dbuser, modified_user)
- new_inbounds = {(i.node_id, i.protocol, i.tag) for i in new_user.inbounds}
- user = UserResponse.model_validate(dbuser)
-
- if user.status in [UserStatus.active, UserStatus.on_hold]:
- asyncio.create_task(marznode.operations.update_user(
- user=new_user, new_inbounds=new_inbounds, old_inbounds=old_inbounds)
- )
- else:
- asyncio.create_task(marznode.operations.remove_user(dbuser))
-
- asyncio.get_running_loop().create_task(report.user_updated(
- user=user,
- by=admin))
-
- logger.info(f"User \"{user.username}\" modified")
-
- if user.status != old_status:
- asyncio.get_running_loop().create_task(report.status_change(
- username=user.username,
- status=user.status,
- user=user,
- by=admin))
- logger.info(
- f"User \"{dbuser.username}\" status changed from {old_status} to {user.status}")
-
- return user
-
-
-@app.delete("/api/user/{username}", tags=['User'])
-async def remove_user(username: str,
- db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- Remove a user
- """
- dbuser = crud.get_user(db, username)
- if not (admin.is_sudo or (dbuser.admin and dbuser.admin.username == admin.username)):
- raise HTTPException(status_code=403, detail="You're not allowed")
-
- if not dbuser:
- raise HTTPException(status_code=404, detail="User not found")
-
- await marznode.operations.remove_user(dbuser)
-
- crud.remove_user(db, dbuser)
- db.flush()
-
- asyncio.create_task(
- report.user_deleted(
- username=dbuser.username,
- by=admin
- ))
- logger.info(f"User \"{username}\" deleted")
- return {}
-
-
-@app.post("/api/user/{username}/reset", tags=['User'], response_model=UserResponse)
-async def reset_user_data_usage(username: str,
- db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- Reset user data usage
- """
- 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")
-
- previous_status = db_user.status
- db_user = crud.reset_user_data_usage(db=db, dbuser=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(f"User \"{username}\"'s usage was reset")
-
- return user
-
-
-@app.post("/api/user/{username}/revoke_sub", tags=['User'], response_model=UserResponse)
-async def revoke_user_subscription(username: str,
- db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- Revoke users subscription (Subscription link and proxies)
- """
- 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")
-
- db_user = crud.revoke_user_sub(db=db, dbuser=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(f"User \"{username}\" subscription revoked")
-
- return user
-
-
-@app.get("/api/users", tags=['User'], response_model=UsersResponse)
-def get_users(offset: int = None,
- limit: int = None,
- username: List[str] = Query(None),
- status: UserStatus = None,
- sort: str = None,
- db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- 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}
-
-
-@app.post("/api/users/reset", tags=['User'])
-async def reset_users_data_usage(db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- Reset all users data usage
- """
- if not admin.is_sudo:
- raise HTTPException(status_code=403, detail="You're not allowed")
-
- 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 {}
-
-
-@app.get("/api/user/{username}/usage", tags=['User'], response_model=UserUsagesResponse)
-def get_user_usage(username: str,
- start: str = None,
- end: str = None,
- db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- Get users usage
- """
- dbuser = crud.get_user(db, username)
- if not dbuser:
- raise HTTPException(status_code=404, detail="User not found")
-
- 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": username}
-
-
-@app.put("/api/user/{username}/set-owner", tags=['User'], response_model=UserResponse)
-def set_owner(username: str,
- 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")
-
- dbuser = crud.get_user(db, username)
- if not dbuser:
- 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")
-
- dbuser = crud.set_owner(db, dbuser, new_admin)
- user = UserResponse.model_validate(dbuser)
-
- logger.info(f"{user.username}'s owner successfully set to {admin.username}")
-
- return user
-
-
-@app.delete("/api/users/expired", tags=['User'])
-def delete_expired(passed_time: int,
- db: Session = Depends(get_db),
- admin: Admin = Depends(Admin.get_current)):
- """
- 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)
-
- dbusers = 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 dbusers if user.expire is not None and user.expire <= expiration_threshold]
- if not expired_users:
- raise HTTPException(status_code=404, detail=f'No expired user found.')
-
- for dbuser in expired_users:
- crud.remove_user(db, dbuser)
-
- asyncio.get_running_loop().create_task(
- report.user_deleted(
- username=dbuser.username,
- by=admin
- ))
-
- logger.info(f"User \"{dbuser.username}\" deleted")
-
- return {}
diff --git a/config.py b/config.py
@@ -22,17 +22,10 @@ VITE_BASE_API = f"http://127.0.0.1:{UVICORN_PORT}/api/" \
if DEBUG and config("VITE_BASE_API", default="/api/") == "/api/" \
else config("VITE_BASE_API", default="/api/")
-XRAY_JSON = config("XRAY_JSON", default="./xray_config.json")
XRAY_FALLBACKS_INBOUND_TAG = config("XRAY_FALLBACKS_INBOUND_TAG", cast=str, default="") or config(
"XRAY_FALLBACK_INBOUND_TAG", cast=str, default=""
)
-XRAY_EXECUTABLE_PATH = config("XRAY_EXECUTABLE_PATH", default="/usr/local/bin/xray")
-if not os.path.isfile(XRAY_EXECUTABLE_PATH):
- raise FileNotFoundError("Xray path doesn't exist.")
-XRAY_ASSETS_PATH = config("XRAY_ASSETS_PATH", default="/usr/local/share/xray")
-XRAY_EXCLUDE_INBOUND_TAGS = config("XRAY_EXCLUDE_INBOUND_TAGS", default='').split()
XRAY_SUBSCRIPTION_URL_PREFIX = config("XRAY_SUBSCRIPTION_URL_PREFIX", default="").strip("/")
-XRAY_SUBSCRIPTION_PATH = config("XRAY_SUBSCRIPTION_PATH", default="sub").strip("/")
TELEGRAM_API_TOKEN = config("TELEGRAM_API_TOKEN", default="")
TELEGRAM_ADMIN_ID = config(
@@ -84,7 +77,4 @@ DISABLE_RECORDING_NODE_USAGE = config("DISABLE_RECORDING_NODE_USAGE", cast=bool,
# headers: profile-update-interval, support-url, profile-title
SUB_UPDATE_INTERVAL = config("SUB_UPDATE_INTERVAL", default="12")
SUB_SUPPORT_URL = config("SUB_SUPPORT_URL", default="https://t.me/")
-SUB_PROFILE_TITLE = config("SUB_PROFILE_TITLE", default="Subscription")
-
-# discord webhook log
-DISCORD_WEBHOOK_URL = config("DISCORD_WEBHOOK_URL", default="")
+SUB_PROFILE_TITLE = config("SUB_PROFILE_TITLE", default="Subscription")
\ No newline at end of file
diff --git a/main.py b/main.py
@@ -1,33 +1,7 @@
import asyncio
+from app.marzneshin import main
-from app import app
-from config import (
- DEBUG,
- UVICORN_HOST,
- UVICORN_PORT,
- UVICORN_UDS,
- UVICORN_SSL_CERTFILE,
- UVICORN_SSL_KEYFILE
-)
-from uvicorn import Config, Server
-
-import logging
if __name__ == "__main__":
loop = asyncio.new_event_loop()
- try:
- cfg = Config(
- app=app,
- host=('0.0.0.0' if DEBUG else 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)
- loop.run_until_complete(server.serve())
- except FileNotFoundError: # to prevent error on removing unix sock
- pass
+ loop.run_until_complete(main())