diff --git a/app/main.py b/app/main.py index 13dd5a9..1dd7732 100644 --- a/app/main.py +++ b/app/main.py @@ -6,7 +6,7 @@ from datetime import datetime, timedelta, UTC from hashlib import sha256 from json import loads as json_loads, dumps as json_dumps from os import getenv as env -from os.path import join, dirname +from os.path import join, dirname, exists, isdir, isfile from textwrap import wrap from uuid import uuid4 @@ -20,6 +20,7 @@ from jose.constants import ALGORITHMS from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from starlette.middleware.cors import CORSMiddleware +from starlette.staticfiles import StaticFiles from orm import Origin, Lease, init as db_init, migrate from util import CASetup, PrivateKey, Cert, ProductMapping, load_file @@ -50,6 +51,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) DT_FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ' PRODUCT_MAPPING = ProductMapping(filename=join(dirname(__file__), 'static/product_mapping.json')) @@ -100,6 +102,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, @@ -206,6 +211,25 @@ async def _manage(request: Request): return Response(response, media_type='text/html', status_code=200) +@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 = join(DRIVERS_DIR, directory) + + if not exists(path) and not isfile(path): + return Response(status_code=404, content=f'Resource "{path}" not found!') + + content = [{ + "type": "file" if isfile(f'{path}/{_}') else "folder" if isdir(f'{path}/{_}') else "unknown", + "name": _, + "link": f'/-/static-drivers/{directory}{_}', + } for _ in listdir(path)] + + return Response(content=json_dumps({"directory": path, "content": content}), media_type='application/json', status_code=200) + + @app.get('/-/origins', summary='* Origins') async def _origins(request: Request, leases: bool = False): session = sessionmaker(bind=db)()