nw-captcha/api.py at main · nemesida-waf/nw-captcha · GitHub

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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
#!/usr/bin/env python3

"""
Nemesida WAF CAPTCHA
Copyright (c) 2017-2025 Pentestit LLC - All Rights Reserved
Author: Romanov R.
"""

import json

from fastapi import FastAPI, Request, Form, status
from fastapi.responses import Response, HTMLResponse, StreamingResponse, RedirectResponse
from fastapi.templating import Jinja2Templates

from core import captcha_img_gen
from core import db_load
from core import mclient
from core import memc_prefix
from core import request_preprocessing
from core import unblock_ip

from logger import log

##

app = FastAPI(redirect_slashes=False, ocs_url=None, redoc_url=None, openapi_url=None)
templates = Jinja2Templates(directory='templates')

##

# Load DB file
db = db_load()

##

@app.get('/', response_class=HTMLResponse)
@app.options('/', response_class=HTMLResponse)
async def main(request: Request):
    try:

        # request preprocessing
        r = request_preprocessing(request)

        # CORS issue fix
        hdrs = {
            'Access-Control-Allow-Origin': '*',
            'Access-Control-Allow-Headers': 'x-nwaf-antibot-id, x-nwaf-antibot-validation'
        }

        # response
        if r:

            # OK: request with query
            if r[1]:
                return templates.TemplateResponse(
                    request=request,
                    headers=hdrs,
                    name='index.html',
                    context={
                        'sid': r[0],
                        'args': r[1]
                    }
                )

            # FAIL: no query in request
            else:
                return Response('Incorrect or empty query', status_code=403)

        else:
            return Response(status_code=400)

    except Exception as e:
        log.error('An error occurred in /: {}'.format(e))


@app.get('/captcha')
async def main(request: Request):
    try:
        sid = str(request.query_params.get('sid'))
        r = captcha_img_gen(sid)
        return Response(status_code=400) if r is None else StreamingResponse(r, media_type='image/png', status_code=200)
    except Exception as e:
        log.error('An error occurred in /captcha: {}'.format(e))


@app.post('/verify', response_class=HTMLResponse)
async def main(request: Request, sid: str = Form(...), answer: str = Form(...)):

    try:

        # receive data from the memcached
        memout = mclient.get(memc_prefix + sid)
        if not memout:

            # request preprocessing
            r = request_preprocessing(request)

            # response
            if r:
                return templates.TemplateResponse('index.html', {'request': request, 'sid': r[0], 'args': r[1], 'status': 0})
            else:
                return Response(status_code=400)

        # success validation
        if str(answer) == str(json.loads(memout).get('answer')):
            tdata = json.loads(memout).get('data', {})
            url = unblock_ip(db, tdata)
            return RedirectResponse(url, status_code=status.HTTP_303_SEE_OTHER) if url else Response(content='', status_code=502)

        # fail validation
        else:

            # request preprocessing
            r = request_preprocessing(request)

            # response
            if r:
                return templates.TemplateResponse('index.html', {'request': request, 'sid': r[0], 'args': r[1], 'status': 1})
            else:
                return Response(status_code=400)

    except Exception as e:
        log.error('An error occurred in /verify: {}'.format(e))