1import asyncio
2import logging
3from datetime import datetime, timedelta
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.db.models import Service
15from app.dependencies import (
16 DBDep,
17 AdminDep,
18 SudoAdminDep,
19 UserDep,
20 StartDateDep,
21 EndDateDep,
22 ModifyUsersAccess,
23)
24from app.models.service import ServiceResponse
25from app.models.user import (
26 UserCreate,
27 UserModify,
28 UserResponse,
29 UserUsagesResponse,
30)
31from app.utils import report
32
33logger = logging.getLogger(__name__)
34router = APIRouter(prefix="/users", tags=["User"])
35
36
37class UsersSortingOptions(str, Enum):
38 USERNAME = "username"
39 USED_TRAFFIC = "used_traffic"
40 DATA_LIMIT = "data_limit"
41 EXPIRE_DATE = "expire_date"
42 CREATED_AT = "created_at"
43
44
45user_filters = [
46 "username",
47 "is_active",
48 "activated",
49 "expired",
50 "data_limit_reached",
51 "enabled",
52 "owner_username",
53]
54
55
56@router.get("", response_model=Page[UserResponse])
57def get_users(
58 db: DBDep,
59 admin: AdminDep,
60 username: list[str] = Query(None),
61 order_by: UsersSortingOptions = Query(None),
62 descending: bool = Query(False),
63 is_active: bool | None = Query(None),
64 activated: bool | None = Query(None),
65 expired: bool | None = Query(None),
66 data_limit_reached: bool | None = Query(None),
67 enabled: bool | None = Query(None),
68 owner_username: str | None = Query(None),
69):
70 """
71 Filters users based on the options
72 """
73 dbadmin = crud.get_admin(db, admin.username)
74
75 query = db.query(User).filter(User.removed == False)
76
77 admin = dbadmin if not admin.is_sudo else None
78
79 for name in user_filters:
80 value = locals().get(name)
81 if value is not None:
82 if name == "username":
83 if len(username) > 1:
84 query = query.filter(User.username.in_(username))
85 else:
86 query = query.filter(
87 User.username.ilike(f"%{username[0]}%")
88 )
89 elif name == "owner_username":
90 if not dbadmin.is_sudo:
91 raise HTTPException(403, "You're not allowed.")
92 filter_admin = crud.get_admin(db, value)
93 if not filter_admin:
94 raise HTTPException(404, "owner_username not found.")
95 query = query.filter(User.admin_id == filter_admin.id)
96 else:
97 query = query.filter(getattr(User, name) == value)
98
99 if admin:
100 query = query.filter(User.admin == admin)
101
102 if order_by:
103 order_column = getattr(User, order_by)
104 if descending:
105 order_column = order_column.desc()
106 query = query.order_by(order_column)
107
108 return paginate(db, query)
109
110
111@router.post("", response_model=UserResponse)
112async def add_user(new_user: UserCreate, db: DBDep, admin: AdminDep):
113 """
114 Add a new user
115
116 - **username** must have 3 to 32 characters and is allowed to contain a-z, 0-9, and underscores in between
117 - **expire_date** must be a datetime
118 - **data_limit** must be in Bytes, e.g. 1073741824B = 1GB
119 - **services** list of service ids
120 """
121
122 try:
123 db_user = crud.create_user(
124 db,
125 new_user,
126 admin=crud.get_admin(db, admin.username),
127 allowed_services=(
128 admin.service_ids
129 if not admin.is_sudo and not admin.all_services_access
130 else None
131 ),
132 )
133 except sqlalchemy.exc.IntegrityError:
134 db.rollback()
135 raise HTTPException(status_code=409, detail="User already exists")
136
137 user = UserResponse.model_validate(db_user)
138 marznode.operations.update_user(user=db_user)
139 asyncio.ensure_future(
140 report.user_created(user=user, user_id=db_user.id, by=admin)
141 )
142 logger.info("New user `%s` added", db_user.username)
143 return user
144
145
146@router.post("/reset")
147async def reset_users_data_usage(db: DBDep, admin: SudoAdminDep):
148 """
149 Reset all users data usage,
150 You will need to restart for this to take effect for now
151 """
152 dbadmin = crud.get_admin(db, admin.username)
153 crud.reset_all_users_data_usage(db=db, admin=dbadmin)
154 return {}
155
156
157@router.delete("/expired")
158async def delete_expired(
159 passed_time: int,
160 db: DBDep,
161 admin: AdminDep,
162 modify_access: ModifyUsersAccess,
163):
164 """
165 Delete expired users
166 - **passed_time** must be a timestamp
167 - This function will delete all expired users that meet the specified number of days passed and can't be undone.
168 """
169
170 dbadmin = crud.get_admin(db, admin.username)
171
172 db_users = crud.get_users(
173 db=db,
174 expired=True,
175 admin=dbadmin if not admin.is_sudo else None,
176 )
177
178 current_time = datetime.utcnow()
179 expiration_threshold = current_time - timedelta(seconds=passed_time)
180 expired_users = [
181 user
182 for user in db_users
183 if user.expire_date is not None
184 and user.expire_date <= expiration_threshold
185 ]
186 if not expired_users:
187 raise HTTPException(status_code=404, detail="No expired user found.")
188
189 for db_user in expired_users:
190 crud.remove_user(db, db_user)
191
192 asyncio.ensure_future(
193 report.user_deleted(username=db_user.username, by=admin)
194 )
195
196 logger.info("User `%s` removed", db_user.username)
197
198 return {}
199
200
201@router.get("/{username}", response_model=UserResponse)
202def get_user(db_user: UserDep):
203 """
204 Get users information
205 """
206 return db_user
207
208
209@router.put("/{username}", response_model=UserResponse)
210async def modify_user(
211 db_user: UserDep,
212 modifications: UserModify,
213 db: DBDep,
214 admin: AdminDep,
215 modify_access: ModifyUsersAccess,
216):
217 """
218 Modify a user
219
220 - set **data_limit** to 0 to make the user unlimited in data, null for no change
221 """
222 active_before = db_user.is_active
223
224 old_inbounds = {(i.node_id, i.protocol, i.tag) for i in db_user.inbounds}
225 new_user = crud.update_user(
226 db,
227 db_user,
228 modifications,
229 allowed_services=(
230 admin.service_ids
231 if not admin.is_sudo and not admin.all_services_access
232 else None
233 ),
234 )
235 active_after = new_user.is_active
236 new_inbounds = {(i.node_id, i.protocol, i.tag) for i in new_user.inbounds}
237
238 inbound_change = old_inbounds != new_inbounds
239
240 if (
241 inbound_change and new_user.is_active
242 ) or active_before != active_after:
243 marznode.operations.update_user(
244 new_user, old_inbounds, remove=not db_user.is_active
245 )
246 db_user.activated = db_user.is_active
247 db.commit()
248
249 asyncio.ensure_future(
250 report.user_updated(
251 user=UserResponse.model_validate(db_user), by=admin
252 )
253 )
254
255 logger.info("User `%s` modified", db_user.username)
256
257 if active_before != active_after:
258 asyncio.ensure_future(
259 report.status_change(
260 username=db_user.username,
261 activation=db_user.activated,
262 user=UserResponse.model_validate(db_user),
263 by=admin,
264 )
265 )
266
267 logger.info(
268 "User `%s` activation changed from `%s` to `%s`",
269 db_user.username,
270 active_before,
271 active_after,
272 )
273
274 return db_user
275
276
277@router.delete("/{username}")
278async def remove_user(
279 db_user: UserDep,
280 db: DBDep,
281 admin: AdminDep,
282 modify_access: ModifyUsersAccess,
283):
284 """
285 Remove a user
286 """
287 marznode.operations.update_user(db_user, remove=True)
288
289 crud.remove_user(db, db_user)
290 db.flush()
291
292 asyncio.ensure_future(
293 report.user_deleted(username=db_user.username, by=admin)
294 )
295 logger.info("User %s deleted", db_user.username)
296 return {}
297
298
299@router.get("/{username}/services", response_model=Page[ServiceResponse])
300def get_user_services(user: UserDep, db: DBDep, admin: AdminDep):
301 """
302 Get user services
303 """
304
305 query = (
306 db.query(Service)
307 .join(Service.users)
308 .where(User.username == user.username)
309 )
310
311 if not admin.is_sudo and not admin.all_services_access:
312 query.filter(Service.id.in_(admin.service_ids))
313
314 return paginate(query)
315
316
317@router.post("/{username}/reset", response_model=UserResponse)
318async def reset_user_data_usage(
319 db_user: UserDep,
320 db: DBDep,
321 admin: AdminDep,
322 modify_access: ModifyUsersAccess,
323):
324 """
325 Reset user data usage
326 """
327 was_active = db_user.is_active
328 db_user = crud.reset_user_data_usage(db, db_user)
329
330 if db_user.is_active and not was_active:
331 marznode.operations.update_user(db_user)
332 db_user.activated = True
333 db.commit()
334
335 user = UserResponse.model_validate(db_user)
336 asyncio.ensure_future(report.user_data_usage_reset(user=user, by=admin))
337
338 logger.info("User `%s`'s usage was reset", db_user.username)
339
340 return user
341
342
343@router.post("/{username}/enable", response_model=UserResponse)
344async def enable_user(
345 db_user: UserDep,
346 db: DBDep,
347 admin: AdminDep,
348 modify_access: ModifyUsersAccess,
349):
350 """
351 Enables a user
352 """
353 if db_user.enabled:
354 raise HTTPException(409, "User is already enabled")
355 db_user.enabled = True
356 db_user.activated = True
357 db.commit()
358
359 marznode.operations.update_user(db_user)
360
361 user = UserResponse.model_validate(db_user)
362
363 logger.info("User `%s` has been enabled", db_user.username)
364
365 return user
366
367
368@router.post("/{username}/disable", response_model=UserResponse)
369async def disable_user(
370 db_user: UserDep,
371 db: DBDep,
372 admin: AdminDep,
373 modify_access: ModifyUsersAccess,
374):
375 """
376 Disables a user
377 """
378 if not db_user.enabled:
379 raise HTTPException(409, "User is not enabled")
380 db_user.enabled = False
381 db_user.activated = False
382 db.commit()
383
384 marznode.operations.update_user(db_user, remove=True)
385
386 user = UserResponse.model_validate(db_user)
387
388 logger.info("User `%s` has been disabled", db_user.username)
389
390 return user
391
392
393@router.post("/{username}/revoke_sub", response_model=UserResponse)
394async def revoke_user_subscription(
395 db_user: UserDep,
396 db: DBDep,
397 admin: AdminDep,
398 modify_access: ModifyUsersAccess,
399):
400 """
401 Revoke users subscription (Subscription link and proxies)
402 """
403 db_user = crud.revoke_user_sub(db, db_user)
404
405 if db_user.is_active:
406 marznode.operations.update_user(db_user, remove=True)
407 marznode.operations.update_user(db_user)
408 user = UserResponse.model_validate(db_user)
409 asyncio.ensure_future(
410 report.user_subscription_revoked(user=user, by=admin)
411 )
412
413 logger.info("User %s subscription revoked", db_user.username)
414
415 return user
416
417
418@router.get("/{username}/usage", response_model=UserUsagesResponse)
419def get_user_usage(
420 db: DBDep, db_user: UserDep, start_date: StartDateDep, end_date: EndDateDep
421):
422 """
423 Get users usage
424 """
425 usages = crud.get_user_usages(db, db_user, start_date, end_date)
426
427 return {"usages": usages, "username": db_user.username}
428
429
430@router.put("/{username}/set-owner", response_model=UserResponse)
431def set_owner(
432 username: str, admin_username: str, db: DBDep, admin: SudoAdminDep
433):
434 db_user = crud.get_user(db, username)
435 if not db_user:
436 raise HTTPException(status_code=404, detail="User not found")
437
438 new_admin = crud.get_admin(db, username=admin_username)
439 if not new_admin:
440 raise HTTPException(status_code=404, detail="Admin not found")
441
442 db_user = crud.set_owner(db, db_user, new_admin)
443 user = UserResponse.model_validate(db_user)
444
445 logger.info(
446 "`%s`'s owner successfully set to `%s`", user.username, admin.username
447 )
448
449 return user