sqlmap/lib/utils/httpd.py at 72fcb66fe8ad3bc848d792bcce44ac30a31c9575 · turkdevops/sqlmap · 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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
#!/usr/bin/env python

"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission
"""

from __future__ import print_function

import mimetypes
import gzip
import os
import re
import sys
import threading
import time
import traceback

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))

from lib.core.enums import HTTP_HEADER
from lib.core.settings import UNICODE_ENCODING
from lib.core.settings import VERSION_STRING
from thirdparty import six
from thirdparty.six.moves import BaseHTTPServer as _BaseHTTPServer
from thirdparty.six.moves import http_client as _http_client
from thirdparty.six.moves import socketserver as _socketserver
from thirdparty.six.moves import urllib as _urllib

HTTP_ADDRESS = "0.0.0.0"
HTTP_PORT = 8951
DEBUG = True
HTML_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "data", "html"))
DISABLED_CONTENT_EXTENSIONS = (".py", ".pyc", ".md", ".txt", ".bak", ".conf", ".zip", "~")

class ThreadingServer(_socketserver.ThreadingMixIn, _BaseHTTPServer.HTTPServer):
    def finish_request(self, *args, **kwargs):
        try:
            _BaseHTTPServer.HTTPServer.finish_request(self, *args, **kwargs)
        except Exception:
            if DEBUG:
                traceback.print_exc()

class ReqHandler(_BaseHTTPServer.BaseHTTPRequestHandler):
    def do_GET(self):
        path, query = self.path.split('?', 1) if '?' in self.path else (self.path, "")
        params = {}
        content = None

        if query:
            params.update(_urllib.parse.parse_qs(query))

        for key in params:
            if params[key]:
                params[key] = params[key][-1]

        self.url, self.params = path, params

        if path == '/':
            path = "index.html"

        path = path.strip('/')

        path = path.replace('/', os.path.sep)
        path = os.path.abspath(os.path.join(HTML_DIR, path)).strip()

        if not os.path.isfile(path) and os.path.isfile("%s.html" % path):
            path = "%s.html" % path

        if ".." not in os.path.relpath(path, HTML_DIR) and os.path.isfile(path) and not path.endswith(DISABLED_CONTENT_EXTENSIONS):
            content = open(path, "rb").read()
            self.send_response(_http_client.OK)
            self.send_header(HTTP_HEADER.CONNECTION, "close")
            self.send_header(HTTP_HEADER.CONTENT_TYPE, mimetypes.guess_type(path)[0] or "application/octet-stream")
        else:
            content = ("<!DOCTYPE html><html lang=\"en\"><head><title>404 Not Found</title></head><body><h1>Not Found</h1><p>The requested URL %s was not found on this server.</p></body></html>" % self.path.split('?')[0]).encode(UNICODE_ENCODING)
            self.send_response(_http_client.NOT_FOUND)
            self.send_header(HTTP_HEADER.CONNECTION, "close")

        if content is not None:
            for match in re.finditer(b"<!(\\w+)!>", content):
                name = match.group(1)
                _ = getattr(self, "_%s" % name.lower(), None)
                if _:
                    content = self._format(content, **{name: _()})

            if "gzip" in self.headers.get(HTTP_HEADER.ACCEPT_ENCODING):
                self.send_header(HTTP_HEADER.CONTENT_ENCODING, "gzip")
                _ = six.BytesIO()
                compress = gzip.GzipFile("", "w+b", 9, _)
                compress._stream = _
                compress.write(content)
                compress.flush()
                compress.close()
                content = compress._stream.getvalue()

            self.send_header(HTTP_HEADER.CONTENT_LENGTH, str(len(content)))

        self.end_headers()

        if content:
            self.wfile.write(content)

        self.wfile.flush()

    def _format(self, content, **params):
        if content:
            for key, value in params.items():
                content = content.replace("<!%s!>" % key, value)

        return content

    def version_string(self):
        return VERSION_STRING

    def log_message(self, format, *args):
        return

    def finish(self):
        try:
            _BaseHTTPServer.BaseHTTPRequestHandler.finish(self)
        except Exception:
            if DEBUG:
                traceback.print_exc()

def start_httpd():
    server = ThreadingServer((HTTP_ADDRESS, HTTP_PORT), ReqHandler)
    thread = threading.Thread(target=server.serve_forever)
    thread.daemon = True
    thread.start()

    print("[i] running HTTP server at '%s:%d'" % (HTTP_ADDRESS, HTTP_PORT))

if __name__ == "__main__":
    try:
        start_httpd()

        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        pass