1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
from typing import Annotated
import asyncio
from datetime import datetime, timedelta
from fastapi import FastAPI, Request, Depends
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
from pydantic import BaseModel
from utils import get_avatar_urls, get_client_geo
from app.tasks import restricted_exec_task
templates = Jinja2Templates(directory="templates")
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/", response_class=HTMLResponse)
async def index(
request: Request,
avatar_urls: Annotated[list[str], Depends(get_avatar_urls)],
):
return templates.TemplateResponse(
"index.html",
{"request": request, "avatar_urls": avatar_urls},
)
@app.get("/client_geo", response_class=HTMLResponse)
async def client_addr(
request: Request,
client_geo: Annotated[dict, Depends(get_client_geo)],
):
return templates.TemplateResponse(
"partials/client_geo.html",
{"request": request, "client_geo": client_geo},
)
class RestrictedExecBase(BaseModel):
code: str
@app.post('/restricted_exec')
async def post_restricted_exec(
body: RestrictedExecBase
):
result = restricted_exec_task.delay(body.code)
start_time = datetime.now()
while datetime.now() - start_time < timedelta(seconds=5):
if result.ready():
return result.get()
await asyncio.sleep(0.1)
result.revoke(terminate=True, signal='SIGTERM')
return [
'Execution timeout, task revoked'
]
|
