1import asyncio
2import logging
3from datetime import datetime, timezone
4from enum import Enum
5
6import sqlalchemy
7from fastapi import APIRouter
8from fastapi import HTTPException, Query
9from fastapi_pagination.ext.sqlalchemy import paginate
10from fastapi_pagination.links import Page
11
12from app import marznode
13from app.db import crud, User
14from app.dependencies import (
15 DBDep,
16 AdminDep,
17 SudoAdminDep,
18 UserDep,
19 StartDateDep,
20 EndDateDep,
21)
22from app.models.user import (
23 UserCreate,
24 UserModify,
25 UserResponse,
26 UserStatus,
27 UserUsagesResponse,
28)
29from app.utils import report
30
31logger = logging.getLogger(__name__)
32router = APIRouter(prefix="/users", tags=["User"])
33
34
35class UsersSortingOptions(str, Enum):
36 username = "username"
37 used_traffic = "used_traffic"
38 data_limit = "data_limit"
39 expire = "expire"
40 created_at = "created_at"
41
42
43@router.get("", response_model=Page[UserResponse])
44def get_users(
45 db: DBDep,
46 admin: AdminDep,
47 username: list[str] = Query(None),
48 status: list[UserStatus] = Query(None),
49 order_by: UsersSortingOptions = Query(None),
50 descending: bool = Query(False),
51):
52 """
53 Filters users based on the options
54 """
55 dbadmin = crud.get_admin(db, admin.username)
56
57 query = db.query(User)
58
59 admin = dbadmin if not admin.is_sudo else None
60
61 if username:
62 if len(username) > 1:
63 query = query.filter(User.username.in_(username))
64 else:
65 query = query.filter(User.username.ilike(f"%{username[0]}%"))
66
67 if status:
68 query = query.filter(User.status.in_(status))
69
70 if admin:
71 query = query.filter(User.admin == admin)
72
73 if order_by:
74 order_column = getattr(User, order_by)
75 if descending:
76 order_column = order_column.desc()
77 query = query.order_by(order_column)
78
79 return paginate(db, query)
80
81
82@router.post("", response_model=UserResponse)
83async def add_user(new_user: UserCreate, db: DBDep, admin: AdminDep):
84 """
85 Add a new user
86
87 - **username** must have 3 to 32 characters and is allowed to contain a-z, 0-9, and underscores in between
88 - **expire** must be a UTC timestamp
89 - **data_limit** must be in Bytes, e.g. 1073741824B = 1GB
90 - **services** list of service ids
91 """
92 # TODO expire should be datetime instead of timestamp
93
94 try:
95 db_user = crud.create_user(
96 db, new_user, admin=crud.get_admin(db, admin.username)
97 )
98 except sqlalchemy.exc.IntegrityError:
99 db.rollback()
100 raise HTTPException(status_code=409, detail="User already exists")
101
102 user = UserResponse.model_validate(db_user)
103 await marznode.operations.update_user(user=db_user)
104 asyncio.create_task(report.user_created(user=user, user_id=db_user.id, by=admin))
105 logger.info("New user `%s` added", db_user.username)
106 return user
107
108
109@router.post("/reset")
110async def reset_users_data_usage(db: DBDep, admin: SudoAdminDep):
111 """
112 Reset all users data usage,
113 You will need to restart for this to take effect for now
114 """
115 dbadmin = crud.get_admin(db, admin.username)
116 crud.reset_all_users_data_usage(db=db, admin=dbadmin)
117 return {}
118
119
120@router.delete("/expired")
121def delete_expired(passed_time: int, db: DBDep, admin: AdminDep):
122 """
123 Delete expired users
124 - **passed_time** must be a timestamp
125 - This function will delete all expired users that meet the specified number of days passed and can't be undone.
126 """
127
128 dbadmin = crud.get_admin(db, admin.username)
129
130 db_users = crud.get_users(
131 db=db,
132 status=[UserStatus.expired, UserStatus.limited],
133 admin=dbadmin if not admin.is_sudo else None,
134 )
135
136 current_time = int(datetime.now(timezone.utc).timestamp())
137 expiration_threshold = current_time - passed_time
138 expired_users = [
139 user
140 for user in db_users
141 if user.expire is not None and user.expire <= expiration_threshold
142 ]
143 if not expired_users:
144 raise HTTPException(status_code=404, detail="No expired user found.")
145
146 for db_user in expired_users:
147 crud.remove_user(db, db_user)
148
149 asyncio.create_task(report.user_deleted(username=db_user.username, by=admin))
150
151 logger.info("User `%s` removed", db_user.username)
152
153 return {}
154
155
156@router.get("/{username}", response_model=UserResponse)
157def get_user(db_user: UserDep):
158 """
159 Get users information
160 """
161 return db_user
162
163
164@router.put("/{username}", response_model=UserResponse)
165async def modify_user(
166 db_user: UserDep, modifications: UserModify, db: DBDep, admin: AdminDep
167):
168 """
169 Modify a user
170
171 - set **expire** to 0 to make the user unlimited in time, null for no change
172 - set **data_limit** to 0 to make the user unlimited in data, null for no change
173 """
174 old_status = db_user.status
175 status_change = bool(
176 modifications.status and modifications.status != db_user.status
177 )
178 if status_change and (
179 modifications.status == UserStatus.disabled
180 and old_status in {UserStatus.active, UserStatus.on_hold}
181 ):
182 await marznode.operations.remove_user(db_user)
183
184 old_inbounds = {(i.node_id, i.protocol, i.tag) for i in db_user.inbounds}
185 new_user = crud.update_user(db, db_user, modifications)
186 new_inbounds = {(i.node_id, i.protocol, i.tag) for i in new_user.inbounds}
187 user = UserResponse.model_validate(db_user)
188
189 inbound_change = old_inbounds != new_inbounds
190 if inbound_change and user.status in {UserStatus.active, UserStatus.on_hold}:
191 await marznode.operations.update_user(new_user, old_inbounds)
192
193 asyncio.create_task(report.user_updated(user=user, by=admin))
194
195 logger.info("User `%s` modified", user.username)
196
197 if status_change:
198 asyncio.create_task(
199 report.status_change(
200 username=user.username, status=user.status, user=user, by=admin
201 )
202 )
203
204 logger.info(
205 "User `%s` status changed from `%s` to `%s`",
206 user.username,
207 old_status,
208 user.status,
209 )
210
211 return user
212
213
214@router.delete("/{username}")
215async def remove_user(db_user: UserDep, db: DBDep, admin: AdminDep):
216 """
217 Remove a user
218 """
219 await marznode.operations.remove_user(db_user)
220
221 crud.remove_user(db, db_user)
222 db.flush()
223
224 asyncio.create_task(report.user_deleted(username=db_user.username, by=admin))
225 logger.info("User %s deleted", db_user.username)
226 return {}
227
228
229@router.post("/{username}/reset", response_model=UserResponse)
230async def reset_user_data_usage(db_user: UserDep, db: DBDep, admin: AdminDep):
231 """
232 Reset user data usage
233 """
234 previous_status = db_user.status
235 db_user = crud.reset_user_data_usage(db, db_user)
236
237 if db_user.status == UserStatus.active and previous_status == UserStatus.limited:
238 await marznode.operations.update_user(db_user)
239
240 user = UserResponse.model_validate(db_user)
241 asyncio.create_task(report.user_data_usage_reset(user=user, by=admin))
242
243 logger.info("User `%s`'s usage was reset", db_user.username)
244
245 return user
246
247
248@router.post("/{username}/revoke_sub", response_model=UserResponse)
249async def revoke_user_subscription(db_user: UserDep, db: DBDep, admin: AdminDep):
250 """
251 Revoke users subscription (Subscription link and proxies)
252 """
253 db_user = crud.revoke_user_sub(db, db_user)
254
255 if db_user.status in [UserStatus.active, UserStatus.on_hold]:
256 await marznode.operations.remove_user(db_user)
257 await marznode.operations.update_user(db_user)
258 user = UserResponse.model_validate(db_user)
259 asyncio.create_task(report.user_subscription_revoked(user=user, by=admin))
260
261 logger.info("User %s subscription revoked", db_user.username)
262
263 return user
264
265
266@router.get("/{username}/usage", response_model=UserUsagesResponse)
267def get_user_usage(
268 db: DBDep, db_user: UserDep, start_date: StartDateDep, end_date: EndDateDep
269):
270 """
271 Get users usage
272 """
273 usages = crud.get_user_usages(db, db_user, start_date, end_date)
274
275 return {"usages": usages, "username": db_user.username}
276
277
278@router.put("/{username}/set-owner", response_model=UserResponse)
279def set_owner(username: str, admin_username: str, db: DBDep, admin: SudoAdminDep):
280 db_user = crud.get_user(db, username)
281 if not db_user:
282 raise HTTPException(status_code=404, detail="User not found")
283
284 new_admin = crud.get_admin(db, username=admin_username)
285 if not new_admin:
286 raise HTTPException(status_code=404, detail="Admin not found")
287
288 db_user = crud.set_owner(db, db_user, new_admin)
289 user = UserResponse.model_validate(db_user)
290
291 logger.info("`%s`'s owner successfully set to `%s`", user.username, admin.username)
292
293 return user