47 lines
1.2 KiB
Python
47 lines
1.2 KiB
Python
import os
|
|
|
|
import bcrypt
|
|
from fastapi import Depends, FastAPI, HTTPException, UploadFile, Response
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from dotenv import load_dotenv
|
|
from pathlib import Path
|
|
|
|
load_dotenv()
|
|
|
|
app = FastAPI()
|
|
|
|
UPLOAD_HASH = os.environ["UPLOAD_HASH"]
|
|
UPLOAD_PATH = "uploads"
|
|
|
|
security = HTTPBearer()
|
|
|
|
|
|
def authenticate(
|
|
credentials: HTTPAuthorizationCredentials = Depends(security),
|
|
):
|
|
token = credentials.credentials.encode()
|
|
|
|
if not bcrypt.checkpw(token, UPLOAD_HASH.encode()):
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail="Invalid authentication credentials",
|
|
)
|
|
|
|
|
|
@app.get('/test', dependencies=[Depends(authenticate)])
|
|
async def file_test():
|
|
test_file = f"/{UPLOAD_PATH}/test.txt"
|
|
with open(test_file) as f:
|
|
content = f.read()
|
|
return {'message': content}
|
|
|
|
|
|
@app.post("/upload", dependencies=[Depends(authenticate)])
|
|
async def upload(file: UploadFile):
|
|
with open(Path(f"/{UPLOAD_PATH}/{file.filename}"), "wb") as f:
|
|
while chunk := await file.read(64 * 1024):
|
|
f.write(chunk)
|
|
|
|
await file.close()
|
|
|
|
return {"status": "success"} |