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