diff --git a/app/main.py b/app/main.py index d0a79f6..1f915ce 100644 --- a/app/main.py +++ b/app/main.py @@ -1,11 +1,12 @@ import logging +import os.path from base64 import b64encode as b64enc from calendar import timegm from contextlib import asynccontextmanager from datetime import datetime, timedelta, UTC from hashlib import sha256 from json import loads as json_loads -from os import getenv as env +from os import getenv as env, listdir from os.path import join, dirname from uuid import uuid4 @@ -13,6 +14,7 @@ from dateutil.relativedelta import relativedelta from dotenv import load_dotenv from fastapi import FastAPI from fastapi.requests import Request +from fastapi.staticfiles import StaticFiles from jose import jws, jwk, jwt, JWTError from jose.constants import ALGORITHMS from sqlalchemy import create_engine @@ -50,6 +52,7 @@ LEASE_RENEWAL_PERIOD = float(env('LEASE_RENEWAL_PERIOD', 0.15)) LEASE_RENEWAL_DELTA = timedelta(days=int(env('LEASE_EXPIRE_DAYS', 90)), hours=int(env('LEASE_EXPIRE_HOURS', 0))) CLIENT_TOKEN_EXPIRE_DELTA = relativedelta(years=12) CORS_ORIGINS = str(env('CORS_ORIGINS', '')).split(',') if (env('CORS_ORIGINS')) else [f'https://{DLS_URL}'] +DRIVERS_DIR = env('DRIVERS_DIR', None) jwt_encode_key = jwk.construct(INSTANCE_KEY_RSA.export_key().decode('utf-8'), algorithm=ALGORITHMS.RS256) jwt_decode_key = jwk.construct(INSTANCE_KEY_PUB.export_key().decode('utf-8'), algorithm=ALGORITHMS.RS256) @@ -88,6 +91,9 @@ async def lifespan(_: FastAPI): config = dict(openapi_url=None, docs_url=None, redoc_url=None) # dict(openapi_url='/-/openapi.json', docs_url='/-/docs', redoc_url='/-/redoc') app = FastAPI(title='FastAPI-DLS', description='Minimal Delegated License Service (DLS).', version=VERSION, lifespan=lifespan, **config) +if DRIVERS_DIR is not None: + app.mount('/-/static-drivers', StaticFiles(directory=str(DRIVERS_DIR), html=False), name='drivers') + app.debug = DEBUG app.add_middleware( CORSMiddleware, @@ -186,6 +192,25 @@ async def _manage(request: Request): return HTMLr(response) +@app.get('/-/drivers/{directory:path}', summary='* List drivers directory') +async def _drivers(request: Request, directory: str | None): + if DRIVERS_DIR is None: + return Response(status_code=404, content=f'Variable "DRIVERS_DIR" not set.') + + path = os.path.join(DRIVERS_DIR, directory) + + if not os.path.exists(path) and not os.path.isfile(path): + return Response(status_code=404, content=f'Resource "{path}" not found!') + + content = [{ + "type": "file" if os.path.isfile(f'{path}/{_}') else "folder" if os.path.isdir(f'{path}/{_}') else "unknown", + "name": _, + "link": f'/-/static-drivers/{directory}{_}', + } for _ in listdir(path)] + + return JSONr({"directory": path, "content": content}) + + @app.get('/-/origins', summary='* Origins') async def _origins(request: Request, leases: bool = False): session = sessionmaker(bind=db)()