Merge pull request #36 from recursivezero/feature/RTY-260032

update meta tag

Author: recursivezero

.github/workflows/project-assign.yml

@@ -4,7 +4,7 @@ run-name: Assign project under issue and pull requests
 on:
   issues:
     types: [opened]
-  pull_request:
+  pull_request_target:
     types: [opened, reopened]
 
 permissions:

.gitignore

@@ -5,6 +5,7 @@ __pycache__/
 
 .venv*
 venv/
+.venv-dist/
 env/
 .venv*
 

app/main.py

@@ -3,18 +3,12 @@
 from pathlib import Path
 import logging
 import asyncio
-
 from fastapi import FastAPI, Request
-
 from fastapi.responses import JSONResponse
 from fastapi.staticfiles import StaticFiles
 from starlette.middleware.sessions import SessionMiddleware
-
-# from fastapi.exceptions import RequestValidationError
-# from starlette.exceptions import HTTPException as StarletteHTTPException
 from fastapi.exceptions import HTTPException as FastAPIHTTPException
 from fastapi.templating import Jinja2Templates
-
 from app.routes import ui_router
 from app.utils import db
 from app.utils.cache import cleanup_expired
@@ -115,26 +109,6 @@ async def lifespan(app: FastAPI):
     name="qr",
 )
 
-# -----------------------------
-# Global error handler
-# -----------------------------
-# @app.exception_handler(Exception)
-# async def global_exception_handler(request: Request, exc: Exception):
-#  traceback.print_exc()
-#  return JSONResponse(
-#        status_code=500,
-#        content={"success": False, "error": "INTERNAL_SERVER_ERROR"},
-#    )
-
-
-# @app.exception_handler(404)
-# async def custom_404_handler(request: Request, exc):
-#    return templates.TemplateResponse(
-#        "404.html",
-#        {"request": request},
-#        status_code=404,
-#    )
-
 
 @app.exception_handler(FastAPIHTTPException)
 async def http_exception_handler(request: Request, exc: FastAPIHTTPException):

app/routes.py

@@ -34,11 +34,21 @@
     remove_cache_key,
     rev_cache,
 )
-from app.utils.config import DOMAIN, MAX_RECENT_URLS, CACHE_PURGE_TOKEN, QR_DIR
-from app.utils.helper import generate_code, is_valid_url, sanitize_url, format_date
+from app.utils.config import (
+    DOMAIN,
+    MAX_RECENT_URLS,
+    CACHE_PURGE_TOKEN,
+    QR_DIR,
+)
+from app.utils.helper import (
+    generate_code,
+    sanitize_url,
+    is_valid_url,
+    authorize_url,
+    format_date,
+)
 from app.utils.qr import generate_qr_with_logo
 
-# templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
 templates = Jinja2Templates(directory="app/templates")
 # Routers
 ui_router = APIRouter()
@@ -98,12 +108,18 @@ async def create_short_url(
     qr_type: str = Form("short"),
 ):
     session = request.session
-    original_url = sanitize_url(original_url)
+    original_url = sanitize_url(original_url)  # sanitize the URL input
 
-    if not original_url or not is_valid_url(original_url):
+    if not original_url or not is_valid_url(original_url):  # validate the URL
         session["error"] = "Please enter a valid URL."
         return RedirectResponse("/", status_code=status.HTTP_303_SEE_OTHER)
 
+    if not authorize_url(
+        original_url
+    ):  # authorize the URL based on whitelist/blacklist
+        session["error"] = "This domain is not allowed."
+        return RedirectResponse("/", status_code=status.HTTP_303_SEE_OTHER)
+
     short_code: Optional[str] = get_short_from_cache(original_url)
 
     if not short_code and db.is_connected():
@@ -219,7 +235,6 @@ def redirect_short_ui(short_code: str, background_tasks: BackgroundTasks):
                 set_cache_pair(short_code, original_url)
                 return RedirectResponse(original_url)
 
-    # return PlainTextResponse("Invalid short URL", status_code=404)
     raise HTTPException(status_code=404, detail="Page not found")
 
 
@@ -331,9 +346,13 @@ class ShortenRequest(BaseModel):
 @api_v1.post("/shorten")
 def shorten_api(payload: ShortenRequest):
     original_url = sanitize_url(payload.url)
+
     if not is_valid_url(original_url):
         return JSONResponse(status_code=400, content={"error": "INVALID_URL"})
 
+    if not authorize_url(original_url):
+        return JSONResponse(status_code=400, content={"error": "DOMAIN_NOT_ALLOWED"})
+
     short_code = get_short_from_cache(original_url)
     if not short_code:
         short_code = generate_code()