Ajout authentification OAuth2

.env.template

@@ -0,0 +1,3 @@
+SECRET_KEY=
+USERNAME=
+PASS_HASH=

.gitignore

@@ -2,3 +2,5 @@
 /fabriquedoc.iml
 /out/
 *.DS_Store
+/test_confiance_values.http
+/.env

convert_video.py

@@ -0,0 +1,46 @@
+import logging
+import os
+
+import cv2
+
+
+def convert_video(images_path, output_path, width, height, fps, stilltime):
+    """
+    Convert images in output_path into a mp4 file usine OpenCV.
+    :param images_path:
+    :param output_path:
+    :param images_path:
+    :param width:
+    :param height:
+    :return:
+    """
+
+    # define a frame array
+    frame_array = []
+
+    # list all files in images_path
+    files = [f for f in os.listdir(images_path) if os.path.isfile(os.path.join(images_path, f))]
+    # sort the files
+    files.sort()
+
+    # create a video writer object
+
+    for i in range(len(files)):
+        file = os.path.join(images_path, files[i])
+        logging.log(logging.INFO, f'Converting {file} to mp4')
+        img = cv2.imread(file)
+        for j in range(fps * stilltime):
+            frame_array.append(img)
+
+    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+    video_writer = cv2.VideoWriter(output_path,
+                                   fourcc,
+                                   fps,
+                                   (width, height))
+
+    for i in range(len(frame_array)):
+        # writing to a image array
+        video_writer.write(frame_array[i])
+
+    video_writer.release()
+    logging.log(logging.INFO, f'Finished converting {output_path}')

docker-run.sh

@@ -2,4 +2,4 @@ docker stop fabriquedoc
 docker rm fabriquedoc
 docker network create fabriquedoc
 # Ce programme sert à lancer le job_dispatcher dans un docker localement pour tester
-docker run -p 8000:8000 --name fabriquedoc --network fabriquedoc local/fabriquedoc
+docker run -p 8000:8000 --env-file .env --name fabriquedoc --network fabriquedoc local/fabriquedoc

main.py

@@ -16,71 +16,141 @@
     along with this program.  If not, see <https://www.gnu.org/licenses/>. 
 """
 
-import datetime
+from datetime import datetime, timedelta, timezone
 import logging
+from typing import Annotated
 
-from fastapi import FastAPI, UploadFile
+import jwt
+from fastapi import FastAPI, UploadFile, Depends, HTTPException
 from fastapi.responses import FileResponse
 import pypandoc
 import json
+
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
 from fastapi.testclient import TestClient
 import os
 import shutil
-import cv2
+
+from passlib.exc import InvalidTokenError
+from starlette import status
 
 from DocumentSpecs import DocumentSpecs
 from FormatParameters import FormatParameters
 from convert_pdf import convert_pdf
+from convert_video import convert_video
 from extract_emojis import replace_emojis
 from list_dir import list_dir
+from models import UserInDB, User, TokenData, Token
 from responses import Styles, Formats, App
 
+from passlib.context import CryptContext
 
-def convert_video(images_path, output_path, width, height, fps, stilltime):
-    """
-    Convert images in output_path into a mp4 file usine OpenCV.
-    :param images_path:
-    :param output_path:
-    :param images_path:
-    :param width:
-    :param height:
-    :return:
-    """
-
-    # define a frame array
-    frame_array = []
-
-    # list all files in images_path
-    files = [f for f in os.listdir(images_path) if os.path.isfile(os.path.join(images_path, f))]
-    # sort the files
-    files.sort()
-
-    # create a video writer object
-
-    for i in range(len(files)):
-        file = os.path.join(images_path, files[i])
-        logging.log(logging.INFO, f'Converting {file} to mp4')
-        img = cv2.imread(file)
-        for j in range(fps * stilltime):
-            frame_array.append(img)
-
-    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
-    video_writer = cv2.VideoWriter(output_path,
-                                   fourcc,
-                                   fps,
-                                   (width, height))
-
-    for i in range(len(frame_array)):
-        # writing to a image array
-        video_writer.write(frame_array[i])
-
-    video_writer.release()
-    logging.log(logging.INFO, f'Finished converting {output_path}')
+SECRET_KEY = os.getenv("SECRET_KEY")
+USERNAME = os.getenv("USERNAME")
+PASS_HASH = os.getenv("PASS_HASH")
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7  # 7 days
 
+fake_users_db = {
+    "francois": {
+        "username": f"{USERNAME}",
+        "hashed_password": f"{PASS_HASH}",
+        "disabled": False,
+    }
+}
 
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 app = FastAPI()
 
 
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password):
+    return pwd_context.hash(password)
+
+
+def get_user(db, username: str):
+    if username in db:
+        user_dict = db[username]
+        return UserInDB(**user_dict)
+
+
+def authenticate_user(fake_db, username: str, password: str):
+    user = get_user(fake_db, username)
+    if not user:
+        return False
+    if not verify_password(password, user.hashed_password):
+        return False
+    return user
+
+
+def create_access_token(data: dict, expires_delta: timedelta | None = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.now(timezone.utc) + expires_delta
+    else:
+        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: str = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = TokenData(username=username)
+    except InvalidTokenError:
+        raise credentials_exception
+    user = get_user(fake_users_db, username=token_data.username)
+    if user is None:
+        raise credentials_exception
+    return user
+
+
+async def get_current_active_user(
+        current_user: Annotated[User, Depends(get_current_user)],
+):
+    if current_user.disabled:
+        raise HTTPException(status_code=400, detail="Inactive user")
+    return current_user
+
+
+@app.post("/token")
+async def login_for_access_token(
+        form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
+) -> Token:
+    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = create_access_token(
+        data={"sub": user.username}, expires_delta=access_token_expires
+    )
+    return Token(access_token=access_token, token_type="bearer")
+
+
+@app.get("/users/me")
+async def read_users_me(
+        current_user: Annotated[User, Depends(get_current_active_user)],
+):
+    return current_user
+
+
 @app.get("/")
 async def get_root():
     app = App(app='fabriquedoc')
@@ -88,19 +158,19 @@ async def get_root():
 
 
 @app.get("/styles/")
-async def get_styles():
+async def get_styles(current_user: Annotated[User, Depends(get_current_active_user)]):
     styles = Styles(styles=list_dir("./styles"))
     return styles
 
 
 @app.get("/formats/{style}/")
-async def get_formats(style: str):
+async def get_formats(style: str, current_user: Annotated[User, Depends(get_current_active_user)]):
     formats = Formats(formats=list_dir(f"./styles/{style}/"))
     return formats
 
 
 @app.get("/format_parameters/{style}/{format}/")
-async def get_format_parameters(style: str, format: str):
+async def get_format_parameters(style: str, format: str, current_user: Annotated[User, Depends(get_current_active_user)]):
     # open styles/format_parameters.json as a dictionary
     with open(f"./styles/{style}/format_parameters.json", "r") as f:
         format_data = json.load(f).get(format)
@@ -111,7 +181,7 @@ async def get_format_parameters(style: str, format: str):
 
 
 @app.get("/images/")
-async def get_images():
+async def get_images(current_user: Annotated[User, Depends(get_current_active_user)]):
     # list all files in resources/images
     files = [f for f in os.listdir("./resources/images") if os.path.isfile(os.path.join("./resources/images", f))]
     # sort the files
@@ -120,12 +190,12 @@ async def get_images():
 
 
 @app.get("/images/{nom_image}")
-async def get_image(nom_image: str):
+async def get_image(nom_image: str, current_user: Annotated[User, Depends(get_current_active_user)]):
     return FileResponse(f"./resources/images/{nom_image}")
 
 
 @app.post("/images/")
-async def ajouter_image(file: UploadFile):
+async def ajouter_image(file: UploadFile, current_user: Annotated[User, Depends(get_current_active_user)]):
     """
     Add an image to the images folder.
     :param file:
@@ -145,7 +215,7 @@ async def ajouter_image(file: UploadFile):
 
 
 @app.delete("/images/{nom_image}")
-async def supprimer_image(nom_image: str):
+async def supprimer_image(nom_image: str, current_user: Annotated[User, Depends(get_current_active_user)]):
     """
     Delete an image from the images folder.
     :param nom_image:
@@ -161,10 +231,10 @@ async def supprimer_image(nom_image: str):
 
 
 @app.get("/generer/")
-async def generer(specs: DocumentSpecs):
+async def generer(specs: DocumentSpecs, current_user: Annotated[User, Depends(get_current_active_user)]):
     header_file = f'{os.getcwd()}/styles/{specs.style}/{specs.format}/header.tex'
     cover_file = f'{os.getcwd()}/styles/{specs.style}/{specs.format}/cover.tex'
-    datef = datetime.datetime.now().strftime("%Y-%m-%d")
+    datef = datetime.now().strftime("%Y-%m-%d")
     os.makedirs("out", exist_ok=True)
     filters = ['latex-emoji.lua', 'centered.lua']
     pdoc_args = [

models.py

@@ -0,0 +1,19 @@
+from pydantic import BaseModel
+
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+
+class TokenData(BaseModel):
+    username: str | None = None
+
+
+class User(BaseModel):
+    username: str
+    disabled: bool | None = None
+
+
+class UserInDB(User):
+    hashed_password: str

requirements.txt

@@ -1,6 +1,12 @@
-emoji~=2.12.1
-fastapi~=0.111.0
-opencv-python~=4.10.0.84
-pydantic~=2.7.4
-pypandoc~=1.13
+emoji==2.12.1
+fastapi==0.112.1
+opencv_python==4.10.0.84
+passlib==1.7.4
+pydantic==2.8.2
+PyJWT==2.9.0
+pypandoc==1.13
+starlette==0.38.2
+uvicorn==0.30.6
 Wand==0.6.13
+httpx==0.27.0
+python-multipart==0.0.9

test_confiance.http

@@ -1,8 +1,16 @@
 # Test your FastAPI endpoints
 
+POST http://127.0.0.1:8000/token/
+Content-Type: application/x-www-form-urlencoded
+
+username=USERNAME&password=PASSWORD
+
+###
+
 GET http://127.0.0.1:8000/generer/
 Content-Type: application/json
 Accept: application/zip
+Authorization: Bearer TOKEN_HERE
 
 {
   "format": "instagram-fullscreen",
@@ -27,6 +35,7 @@ Accept: application/zip
 GET http://127.0.0.1:8000/generer/
 Content-Type: application/json
 Accept: video/*
+Authorization: Bearer TOKEN_HERE
 
 {
   "format": "instagram-story",
@@ -51,6 +60,7 @@ Accept: video/*
 GET http://127.0.0.1:8000/generer/
 Content-Type: application/json
 Accept: application/pdf
+Authorization: Bearer TOKEN_HERE
 
 {
   "format": "slide169",
@@ -75,6 +85,7 @@ Accept: application/pdf
 GET http://127.0.0.1:8000/generer/
 Content-Type: application/json
 Accept: application/pdf
+Authorization: Bearer TOKEN_HERE
 
 {
   "format": "lettre",
@@ -99,6 +110,7 @@ Accept: application/pdf
 GET http://127.0.0.1:8000/generer/
 Content-Type: application/json
 Accept: application/pdf
+Authorization: Bearer TOKEN_HERE
 
 {
   "format": "instagram-carre",