1from typing import Optional, Annotated
2
3import sqlalchemy
4from fastapi import APIRouter
5from fastapi import Depends, HTTPException, status
6from fastapi.security import OAuth2PasswordRequestForm
7from fastapi_pagination import Page
8from fastapi_pagination.ext.sqlalchemy import paginate
9
10from app.db import Session, crud
11from app.db.models import Admin as DBAdmin, Service, User
12from app.dependencies import AdminDep, SudoAdminDep, DBDep
13from app.marznode.operations import update_user
14from app.models.admin import (
15 Admin,
16 AdminCreate,
17 AdminInDB,
18 Token,
19 AdminPartialModify,
20 AdminResponse,
21)
22from app.models.service import ServiceResponse
23from app.models.user import UserResponse
24from app.utils.auth import create_admin_token
25
26router = APIRouter(tags=["Admin"], prefix="/admins")
27
28
29def authenticate_admin(
30 db: Session, username: str, password: str
31) -> Optional[Admin]:
32 dbadmin = crud.get_admin(db, username)
33 if not dbadmin:
34 return None
35
36 return (
37 dbadmin
38 if AdminInDB.model_validate(dbadmin).verify_password(password)
39 else None
40 )
41
42
43@router.get("", response_model=Page[AdminResponse])
44def get_admins(db: DBDep, admin: SudoAdminDep, username: str | None = None):
45 query = db.query(DBAdmin)
46 if username:
47 query = query.filter(DBAdmin.username.ilike(f"%{username}%"))
48 return paginate(db, query)
49
50
51@router.post("", response_model=Admin)
52def create_admin(new_admin: AdminCreate, db: DBDep, admin: SudoAdminDep):
53 try:
54 dbadmin = crud.create_admin(db, new_admin)
55 except sqlalchemy.exc.IntegrityError:
56 db.rollback()
57 raise HTTPException(status_code=409, detail="Admin already exists")
58
59 return dbadmin
60
61
62@router.get("/current", response_model=Admin)
63def get_current_admin(admin: AdminDep):
64 return admin
65
66
67@router.post("/token", response_model=Token)
68def admin_token(
69 form_data: Annotated[OAuth2PasswordRequestForm, Depends()], db: DBDep
70):
71 if dbadmin := authenticate_admin(
72 db, form_data.username, form_data.password
73 ):
74 return Token(
75 is_sudo=dbadmin.is_sudo,
76 access_token=create_admin_token(
77 form_data.username, is_sudo=dbadmin.is_sudo
78 ),
79 )
80
81 raise HTTPException(
82 status_code=status.HTTP_401_UNAUTHORIZED,
83 detail="Incorrect username or password",
84 headers={"WWW-Authenticate": "Bearer"},
85 )
86
87
88@router.get("/{username}", response_model=AdminResponse)
89def get_admin(
90 username: str,
91 db: DBDep,
92 admin: SudoAdminDep,
93):
94 dbadmin = crud.get_admin(db, username)
95 if not dbadmin:
96 raise HTTPException(status_code=404, detail="Admin not found")
97 return dbadmin
98
99
100@router.put("/{username}", response_model=AdminResponse)
101def modify_admin(
102 username: str,
103 modified_admin: AdminPartialModify,
104 db: DBDep,
105 admin: SudoAdminDep,
106):
107 dbadmin = crud.get_admin(db, username)
108 if not dbadmin:
109 raise HTTPException(status_code=404, detail="Admin not found")
110
111 # If a sudoer admin wants to edit another sudoer
112 if username != admin.username and dbadmin.is_sudo:
113 raise HTTPException(
114 status_code=403,
115 detail="You're not allowed to edit another sudoers account. Use marzneshin-cli instead.",
116 )
117
118 dbadmin = crud.update_admin(db, dbadmin, modified_admin)
119 return dbadmin
120
121
122@router.get("/{username}/services", response_model=Page[ServiceResponse])
123def get_admin_services(username: str, db: DBDep, admin: SudoAdminDep):
124 """
125 Get user services
126 """
127 db_admin = crud.get_admin(db, username)
128 if not db_admin:
129 raise HTTPException(status_code=404, detail="Admin not found")
130
131 if db_admin.is_sudo or db_admin.all_services_access:
132 query = db.query(Service)
133 else:
134 query = (
135 db.query(Service)
136 .join(Service.admins)
137 .where(DBAdmin.id == db_admin.id)
138 )
139
140 return paginate(query)
141
142
143@router.get("/{username}/users", response_model=Page[UserResponse])
144def get_admin_users(username: str, db: DBDep, admin: SudoAdminDep):
145 """
146 Get user services
147 """
148 db_admin = crud.get_admin(db, username)
149 if not db_admin:
150 raise HTTPException(status_code=404, detail="Admin not found")
151
152 query = db.query(User).where(User.admin_id == db_admin.id).filter(User.username.isnot(None))
153
154 return paginate(query)
155
156
157@router.post("/{username}/disable_users", response_model=AdminResponse)
158async def disable_users(username: str, db: DBDep, admin: SudoAdminDep):
159 db_admin = crud.get_admin(db, username)
160 if not db_admin:
161 raise HTTPException(status_code=404, detail="Admin not found")
162
163 if db_admin.is_sudo and db_admin.username != admin.username:
164 raise HTTPException(
165 status_code=403,
166 detail="You're not allowed.",
167 )
168
169 for user in crud.get_users(db, admin=db_admin, enabled=True):
170 if user.activated:
171 update_user(user, remove=True)
172 user.enabled = False
173 user.activated = False
174 db.commit()
175
176 return db_admin
177
178
179@router.post("/{username}/enable_users", response_model=AdminResponse)
180async def enable_users(username: str, db: DBDep, admin: SudoAdminDep):
181 db_admin = crud.get_admin(db, username)
182 if not db_admin:
183 raise HTTPException(status_code=404, detail="Admin not found")
184
185 if db_admin.is_sudo and db_admin.username != admin.username:
186 raise HTTPException(
187 status_code=403,
188 detail="You're not allowed.",
189 )
190
191 for user in crud.get_users(db, admin=db_admin, enabled=False):
192 user.enabled = True
193 if user.is_active:
194 update_user(user)
195 user.activated = True
196 db.commit()
197
198 return db_admin
199
200
201@router.delete("/{username}")
202def remove_admin(username: str, db: DBDep, admin: SudoAdminDep):
203 dbadmin = crud.get_admin(db, username)
204 if not dbadmin:
205 raise HTTPException(status_code=404, detail="Admin not found")
206
207 if dbadmin.is_sudo:
208 raise HTTPException(
209 status_code=403,
210 detail="You're not allowed to delete sudoers accounts. Use marzneshin-cli instead.",
211 )
212
213 crud.remove_admin(db, dbadmin)
214 return {}