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
@@ -3,7 +3,16 @@ from sqlalchemy.orm import Session
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
@@ -20,24 +29,6 @@ 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_user",
"get_user_by_id",
@@ -65,7 +56,6 @@ __all__ = [
"delete_notification_reminder",
"GetDB",
- "get_db",
"User",
"System",
diff --git a/app/dependencies.py b/app/dependencies.py
@@ -1,14 +1,20 @@
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, get_db, User
+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)
@@ -48,9 +54,10 @@ def sudo_admin(admin: Annotated[Admin, Depends(get_current_admin)]):
status_code=status.HTTP_403_FORBIDDEN,
detail="Access Denied",
)
+ return admin
-def get_current_user(username: str, key: str, db: Annotated[Session, Depends(get_db)]):
+def get_subscription_user(username: str, key: str, db: Annotated[Session, Depends(get_db)]):
try:
int(key, 16)
except ValueError:
@@ -62,7 +69,21 @@ def get_current_user(username: str, key: str, db: Annotated[Session, Depends(get
return db_user
-SubUserDep = Annotated[User, Depends(get_current_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
+
+
+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)]
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/user.py b/app/models/user.py
@@ -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,13 +190,6 @@ 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:
diff --git a/app/routes/__init__.py b/app/routes/__init__.py
@@ -0,0 +1,3 @@
+import logging
+
+logger = logging.getLogger(__name__)
\ No newline at end of file
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,14 +1,17 @@
from typing import List, Optional, Annotated
import sqlalchemy
-from app import app
+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 app.dependencies import AdminDep, SudoAdminDep, DBDep
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:
@@ -26,7 +29,7 @@ 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)
+@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):
@@ -42,7 +45,7 @@ def admin_token(form_data: Annotated[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: DBDep,
admin: SudoAdminDep):
@@ -55,7 +58,7 @@ 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: DBDep,
@@ -82,7 +85,7 @@ def modify_admin(username: str,
return dbadmin
-@app.delete("/api/admin/{username}", tags=['Admin'])
+@router.delete("/api/admin/{username}")
def remove_admin(username: str,
db: DBDep,
admin: SudoAdminDep):
@@ -100,12 +103,12 @@ def remove_admin(username: str,
return {}
-@app.get("/api/admin", tags=["Admin"], response_model=Admin)
+@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])
+@router.get("/api/admins", response_model=List[Admin])
def get_admins(db: DBDep,
admin: SudoAdminDep,
offset: int = None,
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,22 +1,24 @@
-import asyncio
-# import time
+import logging
from datetime import datetime
from typing import List
import sqlalchemy
+from fastapi import APIRouter, Depends
from fastapi import HTTPException, WebSocket
-
-from app import app, logger, marznode
+from app import marznode
from app.db import crud, get_tls_certificate
+from app.dependencies import DBDep, SudoAdminDep, sudo_admin
from app.marznode import MarzNodeGRPC
from app.models.admin import Admin
from app.models.node import (NodeCreate, NodeModify, NodeResponse,
NodeSettings, NodeStatus, NodesUsageResponse)
-from app.dependencies import DBDep, SudoAdminDep
+
+logger = logging.getLogger(__name__)
+router = APIRouter(tags=['Node'], dependencies=[Depends(sudo_admin)])
-@app.get("/api/node/settings", tags=['Node'], response_model=NodeSettings)
+@router.get("/api/node/settings", response_model=NodeSettings)
def get_node_settings(db: DBDep,
admin: SudoAdminDep):
tls = crud.get_tls_certificate(db)
@@ -26,7 +28,7 @@ def get_node_settings(db: DBDep,
)
-@app.post("/api/node", tags=['Node'], response_model=NodeResponse)
+@router.post("/api/node", response_model=NodeResponse)
async def add_node(new_node: NodeCreate,
db: DBDep,
admin: SudoAdminDep):
@@ -42,11 +44,11 @@ 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: DBDep,
admin: SudoAdminDep):
@@ -57,7 +59,7 @@ def get_node(node_id: int,
return db_node
-@app.websocket("/api/node/{node_id}/logs")
+@router.websocket("/api/node/{node_id}/logs", dependencies=None)
async def node_logs(node_id: int,
websocket: WebSocket,
db: DBDep):
@@ -93,13 +95,13 @@ async def node_logs(node_id: int,
await websocket.send_text(l)
-@app.get("/api/nodes", tags=['Node'], response_model=List[NodeResponse])
+@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: DBDep,
@@ -117,11 +119,11 @@ 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: DBDep,
admin: SudoAdminDep):
@@ -132,7 +134,7 @@ 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: DBDep,
admin: SudoAdminDep):
@@ -143,11 +145,11 @@ 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)
+@router.get("/api/nodes/usage", response_model=NodesUsageResponse)
def get_usage(db: DBDep,
admin: SudoAdminDep,
start: str = None,
diff --git a/app/views/service.py b/app/routes/service.py
rename from app/views/service.py
rename to app/routes/service.py
@@ -1,19 +1,20 @@
from typing import List
import sqlalchemy
+from fastapi import APIRouter, Depends
from fastapi import HTTPException
-from app import app
from app.db import crud
+from app.dependencies import DBDep, sudo_admin
from app.models.service import (ServiceCreate, ServiceModify,
ServiceResponse)
-from app.dependencies import AdminDep, DBDep, SudoAdminDep
+router = APIRouter(dependencies=[Depends(sudo_admin)], tags=['Service'])
-@app.post("/api/service", tags=['Service'], response_model=ServiceResponse)
+
+@router.post("/api/service", response_model=ServiceResponse)
def add_service(new_service: ServiceCreate,
- db: DBDep,
- admin: SudoAdminDep):
+ db: DBDep):
"""
Add a new user template
@@ -29,8 +30,8 @@ def add_service(new_service: ServiceCreate,
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: DBDep, admin: SudoAdminDep):
+@router.get("/api/service/{id}", response_model=ServiceResponse)
+def get_service(id: int, db: DBDep):
"""
Get Service information with id
"""
@@ -41,11 +42,10 @@ def get_service(id: int, db: DBDep, admin: SudoAdminDep):
return dbservice
-@app.put("/api/service/{id}", tags=['Service'], response_model=ServiceResponse)
+@router.put("/api/service/{id}", response_model=ServiceResponse)
def modify_service(id: int,
- modify_service: ServiceModify,
- db: DBDep,
- admin: SudoAdminDep):
+ modification: ServiceModify,
+ db: DBDep):
"""
Modify Service
@@ -54,24 +54,20 @@ def modify_service(id: int,
- **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 dbservice:
raise HTTPException(status_code=404, detail="Service not found")
try:
- return crud.update_service(db, dbservice, modify_service)
+ 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")
-@app.delete("/api/service/{id}", tags=['Service'])
+@router.delete("/api/service/{id}")
def remove_service(id: int,
- db: DBDep,
- admin: SudoAdminDep):
+ db: DBDep):
dbservice = crud.get_service(db, id)
if not dbservice:
raise HTTPException(status_code=404, detail="Service not found")
@@ -79,9 +75,8 @@ def remove_service(id: int,
return crud.remove_service(db, dbservice)
-@app.get("/api/services", tags=['Service'], response_model=List[ServiceResponse])
+@router.get("/api/services", response_model=List[ServiceResponse])
def get_services(db: DBDep,
- admin: AdminDep,
offset: int = None,
limit: int = None):
- return crud.get_services(db) #, offset, limit)
+ 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,15 +1,15 @@
import re
from datetime import datetime
+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 crud
from app.dependencies import DBDep, SubUserDep
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,
@@ -18,6 +18,8 @@ from config import (
XRAY_SUBSCRIPTION_PATH
)
+router = APIRouter(tags=['Subscription'])
+
def get_subscription_user_info(user: UserResponse) -> dict:
return {
@@ -28,8 +30,8 @@ def get_subscription_user_info(user: UserResponse) -> dict:
}
-@app.get("/%s/{username}/{key}/" % XRAY_SUBSCRIPTION_PATH, tags=['Subscription'])
-@app.get("/%s/{username}/{key}" % XRAY_SUBSCRIPTION_PATH, include_in_schema=False)
+@router.get("/%s/{username}/{key}/" % XRAY_SUBSCRIPTION_PATH)
+@router.get("/%s/{username}/{key}" % XRAY_SUBSCRIPTION_PATH, include_in_schema=False)
def user_subscription(db_user: SubUserDep,
request: Request,
db: DBDep,
@@ -42,10 +44,15 @@ def user_subscription(db_user: SubUserDep,
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}
)
)
@@ -85,12 +92,12 @@ def user_subscription(db_user: SubUserDep,
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)
+@router.get("/%s/{username}/{key}/info" % XRAY_SUBSCRIPTION_PATH, response_model=UserResponse)
def user_subscription_info(db_user: SubUserDep):
return db_user
-@app.get("/%s/{username}/{key}/usage" % XRAY_SUBSCRIPTION_PATH, tags=['Subscription'])
+@router.get("/%s/{username}/{key}/usage" % XRAY_SUBSCRIPTION_PATH)
def user_get_usage(db_user: SubUserDep,
db: DBDep,
start: str = None,
@@ -111,7 +118,7 @@ def user_get_usage(db_user: SubUserDep,
return {"usages": usages, "username": db_user.username}
-@app.get("/%s/{username}/{key}/{client_type}" % XRAY_SUBSCRIPTION_PATH, tags=['Subscription'])
+@router.get("/%s/{username}/{key}/{client_type}" % XRAY_SUBSCRIPTION_PATH)
def user_subscription_with_client_type(
db_user: SubUserDep,
request: Request,
diff --git a/app/views/system.py b/app/routes/system.py
rename from app/views/system.py
rename to app/routes/system.py
@@ -1,17 +1,21 @@
from typing import Dict, List, Union
-from app import app
+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, ProxyTypes, InboundBase
+from app.models.proxy import InboundHost, Inbound
from app.models.system import SystemStats
from app.models.user import UserStatus
-from app.dependencies import DBDep, AdminDep, SudoAdminDep
from app.utils.system import memory_usage, cpu_usage, realtime_bandwidth
-from app import __version__
+
+router = APIRouter(tags=["System"])
-@app.get("/api/system", tags=["System"], response_model=SystemStats)
+@router.get("/api/system", response_model=SystemStats)
def get_system_stats(db: DBDep, admin: AdminDep):
mem = memory_usage()
cpu = cpu_usage()
@@ -37,20 +41,20 @@ def get_system_stats(db: DBDep, admin: AdminDep):
)
-@app.get('/api/inbounds', tags=["System"], response_model=List[Inbound])
+@router.get('/api/inbounds', response_model=List[Inbound])
def get_inbounds(db: DBDep, admin: SudoAdminDep):
inbounds = crud.get_all_inbounds(db)
return inbounds
-@app.get('/api/hosts', tags=["System"], response_model=Dict[int, List[InboundHost]])
+@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
-@app.put('/api/hosts', tags=["System"], response_model=Dict[str, List[InboundHost]])
+@router.put('/api/hosts', response_model=Dict[str, List[InboundHost]])
def modify_hosts(modified_hosts: Dict[str, List[InboundHost]],
db: DBDep,
admin: SudoAdminDep):
diff --git a/app/routes/user.py b/app/routes/user.py
@@ -0,0 +1,313 @@
+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
+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 an 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: str = None,
+ end: str = None):
+ """
+ Get users usage
+ """
+ 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, 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/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/user.py b/app/views/user.py
@@ -1,343 +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
-from app.dependencies import get_current_admin, DBDep, AdminDep, SudoAdminDep
-
-
-@app.post("/api/user", tags=['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 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: DBDep,
- admin: AdminDep):
- """
- Get users information
- """
- 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")
-
- return dbuser
-
-
-@app.put("/api/user/{username}", tags=['User'], response_model=UserResponse)
-async def modify_user(username: str,
- modified_user: UserModify,
- db: DBDep,
- admin: AdminDep):
- """
- 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: DBDep,
- admin: AdminDep):
- """
- 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: DBDep,
- admin: AdminDep):
- """
- 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: DBDep,
- admin: AdminDep):
- """
- 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(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}
-
-
-@app.post("/api/users/reset", tags=['User'])
-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 {}
-
-
-@app.get("/api/user/{username}/usage", tags=['User'], response_model=UserUsagesResponse)
-def get_user_usage(db: DBDep,
- admin: AdminDep,
- username: str,
- start: str = None,
- end: str = None):
- """
- 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: DBDep,
- admin: SudoAdminDep):
- 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: 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)
-
- 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/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())