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 return crud.get_admin(db, username)
95
96
97@router.put("/{username}", response_model=AdminResponse)
98def modify_admin(
99 username: str,
100 modified_admin: AdminPartialModify,
101 db: DBDep,
102 admin: SudoAdminDep,
103):
104 dbadmin = crud.get_admin(db, username)
105 if not dbadmin:
106 raise HTTPException(status_code=404, detail="Admin not found")
107
108 # If a sudoer admin wants to edit another sudoer
109 if username != admin.username and dbadmin.is_sudo:
110 raise HTTPException(
111 status_code=403,
112 detail="You're not allowed to edit another sudoers account. Use marzneshin-cli instead.",
113 )
114
115 dbadmin = crud.update_admin(db, dbadmin, modified_admin)
116 return dbadmin
117
118
119@router.get("/{username}/services", response_model=Page[ServiceResponse])
120def get_admin_services(username: str, db: DBDep, admin: SudoAdminDep):
121 """
122 Get user services
123 """
124 db_admin = crud.get_admin(db, username)
125 if not db_admin:
126 raise HTTPException(status_code=404, detail="Admin not found")
127
128 if db_admin.is_sudo or db_admin.all_services_access:
129 query = db.query(Service)
130 else:
131 query = (
132 db.query(Service)
133 .join(Service.admins)
134 .where(DBAdmin.id == db_admin.id)
135 )
136
137 return paginate(query)
138
139
140@router.get("/{username}/users", response_model=Page[UserResponse])
141def get_admin_users(username: str, db: DBDep, admin: SudoAdminDep):
142 """
143 Get user services
144 """
145 db_admin = crud.get_admin(db, username)
146 if not db_admin:
147 raise HTTPException(status_code=404, detail="Admin not found")
148
149 query = db.query(User).where(User.admin_id == db_admin.id)
150
151 return paginate(query)
152
153
154@router.get("/{username}/disable_users", response_model=AdminResponse)
155async def disable_users(username: str, db: DBDep, admin: SudoAdminDep):
156 db_admin = crud.get_admin(db, username)
157 if not db_admin:
158 raise HTTPException(status_code=404, detail="Admin not found")
159
160 if db_admin.is_sudo and db_admin.username != admin.username:
161 raise HTTPException(
162 status_code=403,
163 detail="You're not allowed.",
164 )
165
166 for user in crud.get_users(db, admin=db_admin, enabled=True):
167 if user.activated:
168 update_user(user, remove=True)
169 user.enabled = False
170 user.activated = False
171 db.commit()
172
173 return db_admin
174
175
176@router.get("/{username}/enable_users", response_model=AdminResponse)
177async def enable_users(username: str, db: DBDep, admin: SudoAdminDep):
178 db_admin = crud.get_admin(db, username)
179 if not db_admin:
180 raise HTTPException(status_code=404, detail="Admin not found")
181
182 if db_admin.is_sudo and db_admin.username != admin.username:
183 raise HTTPException(
184 status_code=403,
185 detail="You're not allowed.",
186 )
187
188 for user in crud.get_users(db, admin=db_admin, enabled=False):
189 user.enabled = True
190 if user.is_active:
191 update_user(user)
192 user.activated = True
193 db.commit()
194
195 return db_admin
196
197
198@router.delete("/{username}")
199def remove_admin(username: str, db: DBDep, admin: SudoAdminDep):
200 dbadmin = crud.get_admin(db, username)
201 if not dbadmin:
202 raise HTTPException(status_code=404, detail="Admin not found")
203
204 if dbadmin.is_sudo:
205 raise HTTPException(
206 status_code=403,
207 detail="You're not allowed to delete sudoers accounts. Use marzneshin-cli instead.",
208 )
209
210 crud.remove_admin(db, dbadmin)
211 return {}