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