1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
from fastapi import FastAPI, Path, Depends, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import RedirectResponse
from fastapi.templating import Jinja2Templates
from app.routes.links_route import router as links_router
from app.routes.refresh_route import router as refresh_router
from app.routes.token_route import router as token_router
from typing import Annotated
from fastapi.exceptions import HTTPException
from starlette.status import HTTP_404_NOT_FOUND
from app.util.authentication import get_current_user_from_cookie
from app.schemas.auth_schemas import User
app = FastAPI(
title="LinkLogger API",
version="1.0",
summary="Public API for a combined link shortener and IP logger",
license_info={
"name": "The Unlicense",
"identifier": "Unlicense",
"url": "https://unlicense.org",
},
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
allow_credentials=True,
)
templates = Jinja2Templates(directory="app/templates")
# Import routes
app.include_router(links_router, prefix="/api")
# Must not have a prefix... for some reason you can't change
# the prefix of the Swagger UI OAuth2 redirect to /api/token
# you can only change it to /token, so we have to remove the
# prefix in order to keep logging in via Swagger UI working
app.include_router(token_router)
app.include_router(refresh_router, prefix="/api")
@app.get("/login")
async def login(request: Request):
return templates.TemplateResponse("login.html", {"request": request})
@app.get("/signup")
async def signup(request: Request):
return templates.TemplateResponse("signup.html", {"request": request})
# TODO: Create users routes
# User - register/create
# User - delete
# User - update
# @app.route("/signup", methods=["GET", "POST"])
# def signup():
# if request.method == "POST":
# username = request.form["username"]
# password = request.form["password"]
# # Verify the password meets requirements
# if len(password) < 8:
# return {"status": "Password must be at least 8 characters"}
# if not any(char.isdigit() for char in password):
# return {"status": "Password must contain at least one digit"}
# if not any(char.isupper() for char in password):
# return {
# "status": "Password must contain at least one uppercase letter"
# }
# # Get database session
# db = SessionLocal()
# user = db.query(User).filter(User.username == username).first()
# if user:
# db.close()
# return {"status": "Username not available"}
# # Add information to the database
# hashed_password = bcrypt.hashpw(
# password.encode("utf-8"), bcrypt.gensalt()
# ).decode("utf-8")
# api_key = "".join(
# random.choices(string.ascii_letters + string.digits, k=20)
# )
# new_user = User(
# username=username, password=hashed_password, api_key=api_key
# )
# db.add(new_user)
# db.commit()
# db.close()
# # Log in the newly created user
# flask_user = FlaskUser()
# flask_user.id = username
# login_user(flask_user)
# return {"status": "success"}
# return render_template("signup.html")
@app.get("/dashboard")
async def dashboard(
response: Annotated[
User, RedirectResponse, Depends(get_current_user_from_cookie)
],
request: Request,
):
if isinstance(response, RedirectResponse):
return response
return templates.TemplateResponse(
"dashboard.html", {"request": request, "user": response.username}
)
# @app.get("/{link}")
# async def log_redirect(
# link: Annotated[str, Path(title="Redirect link")],
# request: Request,
# db=Depends(get_db),
# ):
# link = link.upper()
# # If `link` is not exactly 5 characters, return redirect to base url
# if len(link) != 5:
# return RedirectResponse(url="/login")
# # Make sure the link exists in the database
# link_record: Link = db.query(Link).filter(Link.link == link).first()
# if not link_record:
# db.close()
# return RedirectResponse(url="/login")
# else:
# # Log the visit
# if request.headers.get("X-Real-IP"):
# ip = request.headers.get("X-Real-IP").split(",")[0]
# else:
# ip = request.client.host
# user_agent = request.headers.get("User-Agent")
# log(link, ip, user_agent)
# db.close()
# return RedirectResponse(url=link_record.redirect_link)
# Redirect /api -> /api/docs
@app.get("/api")
async def redirect_to_docs():
return RedirectResponse(url="/docs")
# Custom handler for 404 errors
@app.exception_handler(HTTP_404_NOT_FOUND)
async def custom_404_handler(request: Request, exc: HTTPException):
return RedirectResponse(url="/login")
|
