Compare commits
2 Commits
de0a1809a4
...
a04f24fcff
Author | SHA1 | Date | |
---|---|---|---|
a04f24fcff | |||
8c0872ea1a |
@@ -3,12 +3,6 @@ import hmac
|
|||||||
from fastapi import Request
|
from fastapi import Request
|
||||||
from fastapi.exceptions import HTTPException
|
from fastapi.exceptions import HTTPException
|
||||||
|
|
||||||
async def check_ref(request: Request):
|
|
||||||
json = await request.json()
|
|
||||||
if json["ref"] and json["ref"] == f"refs/heads/{getenv('BRANCH')}":
|
|
||||||
return
|
|
||||||
raise HTTPException(status_code=202, detail="Invalid branch")
|
|
||||||
|
|
||||||
async def auth_hook(request: Request):
|
async def auth_hook(request: Request):
|
||||||
try:
|
try:
|
||||||
json = await request.json()
|
json = await request.json()
|
||||||
@@ -38,7 +32,7 @@ async def auth_hook(request: Request):
|
|||||||
|
|
||||||
async def auth_web(request: Request):
|
async def auth_web(request: Request):
|
||||||
token = request._query_params.get("token")
|
token = request._query_params.get("token")
|
||||||
if token is None or token is "":
|
if token == None or token == "":
|
||||||
raise HTTPException(status_code=400, detail="Missing token")
|
raise HTTPException(status_code=400, detail="Missing token")
|
||||||
if token == getenv("TOKEN"):
|
if token == getenv("TOKEN"):
|
||||||
return
|
return
|
||||||
|
@@ -1,13 +1,13 @@
|
|||||||
from fastapi import FastAPI, Request, Depends
|
from fastapi import FastAPI, Request, Depends
|
||||||
from fastapi_responses import custom_openapi
|
from fastapi_responses import custom_openapi
|
||||||
from app.dependencies import auth_hook, auth_web, check_ref
|
from app.dependencies import auth_hook, auth_web
|
||||||
|
|
||||||
app = FastAPI()
|
app = FastAPI()
|
||||||
|
|
||||||
app.openapi = custom_openapi(app)
|
app.openapi = custom_openapi(app)
|
||||||
|
|
||||||
@app.get("/", dependencies=[Depends(auth_web)])
|
@app.get("/", dependencies=[Depends(auth_web)])
|
||||||
@app.post("/", dependencies=[Depends(auth_hook), Depends(check_ref)])
|
@app.post("/", dependencies=[Depends(auth_hook)])
|
||||||
async def hook(req: Request):
|
async def hook(req: Request):
|
||||||
return "Update"
|
return "Update"
|
||||||
|
|
||||||
|
@@ -3,7 +3,7 @@ from fastapi import FastAPI, Request, Depends
|
|||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
import hmac
|
import hmac
|
||||||
from app.main import app
|
from app.main import app
|
||||||
from app.dependencies import auth_hook, auth_web, check_ref
|
from app.dependencies import auth_hook, auth_web
|
||||||
from os import environ, getenv
|
from os import environ, getenv
|
||||||
import json
|
import json
|
||||||
|
|
||||||
@@ -18,10 +18,6 @@ client = TestClient(app)
|
|||||||
async def auth_test_handler(request: Request):
|
async def auth_test_handler(request: Request):
|
||||||
return 200
|
return 200
|
||||||
|
|
||||||
@app.post("/test_ref", dependencies=[Depends(check_ref)])
|
|
||||||
async def auth_test_handler(request: Request):
|
|
||||||
return 200
|
|
||||||
|
|
||||||
@app.get("/test_web", dependencies=[Depends(auth_web)])
|
@app.get("/test_web", dependencies=[Depends(auth_web)])
|
||||||
async def web_test_hnadler(request: Request):
|
async def web_test_hnadler(request: Request):
|
||||||
return 200
|
return 200
|
||||||
@@ -51,14 +47,14 @@ def test_auth():
|
|||||||
assert response.text == '{"detail":"Unauthorized"}'
|
assert response.text == '{"detail":"Unauthorized"}'
|
||||||
|
|
||||||
|
|
||||||
def test_branch():
|
# def test_branch():
|
||||||
payload = {"ref": "refs/heads/master"}
|
# payload = {"ref": "refs/heads/master"}
|
||||||
response = client.post("/test_ref", json= payload)
|
# response = client.post("/test_ref", json= payload)
|
||||||
assert response.status_code == 202
|
# assert response.status_code == 202
|
||||||
|
|
||||||
payload = {"ref": "refs/heads/test"}
|
# payload = {"ref": "refs/heads/test"}
|
||||||
response = client.post("/test_ref", json= payload)
|
# response = client.post("/test_ref", json= payload)
|
||||||
assert response.status_code == 403
|
# assert response.status_code == 403
|
||||||
|
|
||||||
def test_web():
|
def test_web():
|
||||||
response = client.get('/test_web?token={}'.format(getenv("TOKEN")))
|
response = client.get('/test_web?token={}'.format(getenv("TOKEN")))
|
||||||
|
Reference in New Issue
Block a user