1 |
|
from typing import List
|
2 |
|
|
3 |
|
from fastapi import Depends, FastAPI, HTTPException, APIRouter, Form
|
4 |
|
from sqlalchemy.orm import Session
|
5 |
|
from sql_app import crud, models, schemas
|
6 |
1 |
from datetime import datetime
|
7 |
|
from ..database import SessionLocal, engine
|
8 |
|
from fastapi import FastAPI, Request
|
|
2 |
|
|
3 |
from fastapi import Depends, APIRouter, Form
|
|
4 |
from fastapi import Request
|
9 |
5 |
from fastapi.responses import HTMLResponse
|
10 |
|
from fastapi.staticfiles import StaticFiles
|
11 |
6 |
from fastapi.templating import Jinja2Templates
|
|
7 |
from fastapi_jwt_auth import AuthJWT
|
|
8 |
from pydantic import BaseModel
|
|
9 |
from sqlalchemy.orm import Session
|
|
10 |
|
|
11 |
from sql_app import crud, models
|
|
12 |
from ..database import SessionLocal, engine
|
12 |
13 |
|
13 |
14 |
models.Base.metadata.create_all(bind=engine)
|
14 |
15 |
|
... | ... | |
19 |
20 |
device_web = APIRouter(prefix="")
|
20 |
21 |
|
21 |
22 |
|
|
23 |
|
|
24 |
class Settings(BaseModel):
|
|
25 |
authjwt_secret_key: str = "secret"
|
|
26 |
# Configure application to store and get JWT from cookies
|
|
27 |
authjwt_token_location: set = {"cookies"}
|
|
28 |
# Disable CSRF Protection for this example. default is True
|
|
29 |
authjwt_cookie_csrf_protect: bool = False
|
|
30 |
|
|
31 |
|
|
32 |
@AuthJWT.load_config
|
|
33 |
def get_config():
|
|
34 |
return Settings()
|
|
35 |
|
|
36 |
|
|
37 |
fake_users_db = {
|
|
38 |
"admin": {
|
|
39 |
"username": "admin",
|
|
40 |
"password": "admin"
|
|
41 |
},
|
|
42 |
"editor": {
|
|
43 |
"username": "editor",
|
|
44 |
"password": "editor"
|
|
45 |
},
|
|
46 |
}
|
|
47 |
|
|
48 |
|
22 |
49 |
# Dependency
|
23 |
50 |
def get_db():
|
24 |
51 |
db = SessionLocal()
|
... | ... | |
28 |
55 |
db.close()
|
29 |
56 |
|
30 |
57 |
|
|
58 |
@device_web.get("/token", response_class=HTMLResponse)
|
|
59 |
async def login_get(request: Request):
|
|
60 |
return templates.TemplateResponse("login.html", {"request": request})
|
|
61 |
|
|
62 |
|
|
63 |
@device_web.post("/token", response_class=HTMLResponse)
|
|
64 |
async def login(username: str = Form(...), password: str = Form(...), Authorize: AuthJWT = Depends()):
|
|
65 |
user_dict = fake_users_db.get(username)
|
|
66 |
|
|
67 |
access_token = Authorize.create_access_token(subject=username, expires_time=False)
|
|
68 |
refresh_token = Authorize.create_refresh_token(subject=username, expires_time=False)
|
|
69 |
|
|
70 |
# Set the JWT cookies in the response
|
|
71 |
Authorize.set_access_cookies(access_token)
|
|
72 |
Authorize.set_refresh_cookies(refresh_token)
|
|
73 |
return """
|
|
74 |
<html>
|
|
75 |
<head>
|
|
76 |
<title>Some HTML in here</title>
|
|
77 |
</head>
|
|
78 |
<body>
|
|
79 |
<h1>Look ma! HTML!</h1>
|
|
80 |
<form action="/devices-web" method="get">
|
|
81 |
<input type="submit" value="Login" />
|
|
82 |
</form>
|
|
83 |
</body>
|
|
84 |
</html>
|
|
85 |
"""
|
|
86 |
|
|
87 |
|
|
88 |
@device_web.post('/refresh')
|
|
89 |
def refresh(Authorize: AuthJWT = Depends()):
|
|
90 |
Authorize.jwt_refresh_token_required()
|
|
91 |
|
|
92 |
current_user = Authorize.get_jwt_subject()
|
|
93 |
new_access_token = Authorize.create_access_token(subject=current_user)
|
|
94 |
# Set the JWT cookies in the response
|
|
95 |
Authorize.set_access_cookies(new_access_token)
|
|
96 |
return {"msg": "The token has been refresh"}
|
|
97 |
|
|
98 |
|
|
99 |
@device_web.get('/logout', response_class=HTMLResponse)
|
|
100 |
def logout(Authorize: AuthJWT = Depends()):
|
|
101 |
"""
|
|
102 |
Because the JWT are stored in an httponly cookie now, we cannot
|
|
103 |
log the user out by simply deleting the cookies in the frontend.
|
|
104 |
We need the backend to send us a response to delete the cookies.
|
|
105 |
"""
|
|
106 |
Authorize.jwt_required()
|
|
107 |
|
|
108 |
Authorize.unset_jwt_cookies()
|
|
109 |
return """
|
|
110 |
<html>
|
|
111 |
<head>
|
|
112 |
<title>Some HTML in here</title>
|
|
113 |
</head>
|
|
114 |
<body>
|
|
115 |
<h1>Look ma! HTML!</h1>
|
|
116 |
<form action="/devices-web" method="get">
|
|
117 |
<input type="submit" value="Login" />
|
|
118 |
</form>
|
|
119 |
</body>
|
|
120 |
</html>
|
|
121 |
"""
|
|
122 |
|
|
123 |
|
31 |
124 |
@device_web.get("/devices-web", response_class=HTMLResponse)
|
32 |
|
async def read_devices(request: Request, skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
125 |
async def read_devices(request: Request, skip: int = 0, limit: int = 100, db: Session = Depends(get_db),
|
|
126 |
Authorize: AuthJWT = Depends()):
|
33 |
127 |
"""
|
34 |
128 |
Returns template with all devices and its current states
|
35 |
129 |
"""
|
|
130 |
Authorize.jwt_optional()
|
|
131 |
current_user = Authorize.get_jwt_subject()
|
|
132 |
|
36 |
133 |
devices = crud.get_devices(db, skip=skip, limit=limit)
|
37 |
134 |
statuses = []
|
38 |
135 |
# adding state for each device in list
|
39 |
136 |
for i in range(0, len(devices)):
|
40 |
|
statuses.append(devices[i].logs[len(devices[i].logs)-1].status)
|
|
137 |
statuses.append(devices[i].logs[len(devices[i].logs) - 1].status)
|
41 |
138 |
licenses = crud.get_licenses(db, skip=skip, limit=limit)
|
42 |
139 |
return templates.TemplateResponse("devices.html", {"request": request, "devs": len(devices), "devices": devices,
|
43 |
140 |
"statuses": statuses, "licenses": licenses})
|
... | ... | |
63 |
160 |
for i in range(0, len(def_devices)):
|
64 |
161 |
statuses.append(def_devices[i].logs[len(def_devices[i].logs) - 1].status)
|
65 |
162 |
licenses = crud.get_licenses(db, skip=skip, limit=limit)
|
66 |
|
return templates.TemplateResponse("devices.html", {"request": request, "devs": len(def_devices), "devices": def_devices,
|
67 |
|
"statuses": statuses, "licenses": licenses})
|
|
163 |
return templates.TemplateResponse("devices.html",
|
|
164 |
{"request": request, "devs": len(def_devices), "devices": def_devices,
|
|
165 |
"statuses": statuses, "licenses": licenses})
|
68 |
166 |
|
69 |
167 |
|
70 |
168 |
@device_web.get("/device-license/{device_id}", response_class=HTMLResponse)
|
re #9708 Login logic with fastapi-jwt