apiscanner/auth_utils.py at main · perrym/apiscanner · 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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313

########################################################
# APISCAN - API Security Scanner                       #
# Licensed under the AGPL-v3.0                         #
# Author: Perry Mertens pamsniffer@gmail.com (C) 2025  #
# version 4.0 26-04-2026                              #
########################################################

from __future__ import annotations
import logging
import time
import re
import threading
import webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from typing import Optional, Tuple
from urllib.parse import urlparse
import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth

logger = logging.getLogger("auth_utils")


try:
    from requests_ntlm import HttpNtlmAuth
except Exception:
    HttpNtlmAuth = None

try:
    from requests_oauthlib import OAuth2Session
    from oauthlib.oauth2 import BackendApplicationClient, WebApplicationClient
except Exception:
    OAuth2Session = None
    BackendApplicationClient = None
    WebApplicationClient = None


class AuthConfigError(Exception):
    pass

class _CallbackHandler(BaseHTTPRequestHandler):
    _path_with_query: Optional[str] = None
    _event: threading.Event = threading.Event()

    def do_GET(self) -> None:
        type(self)._path_with_query = self.path
        type(self)._event.set()
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.end_headers()
        self.wfile.write(
            b"<html><body><h3>Authentication complete. You may close this window.</h3></body></html>"
        )

    def log_message(self, format, *args) -> None:
        return

# ----------------------- Funtion _start_callback_server ----------------------------#
def _start_callback_server(host: str, port: int) -> Tuple[HTTPServer, threading.Thread]:
    _CallbackHandler._path_with_query = None
    _CallbackHandler._event = threading.Event()

    server = HTTPServer((host, port), _CallbackHandler)
    t = threading.Thread(target=server.serve_forever, daemon=True)
    t.start()
    return server, t


# ----------------------- Funtion _wait_for_callback ----------------------------#
def _wait_for_callback(server: HTTPServer, timeout: int = 300) -> str:
    signalled = _CallbackHandler._event.wait(timeout=float(timeout))
    if not signalled or not _CallbackHandler._path_with_query:
        raise AuthConfigError("Timed out waiting for OAuth2 redirect callback")

    host, port = server.server_address
    return f"http://{host}:{port}{_CallbackHandler._path_with_query}"


# ----------------------- Funtion _apply_api_key ----------------------------#
def _apply_api_key(sess: requests.Session, args) -> None:
    api_key = getattr(args, "apikey", None)
    if api_key:
        api_key = api_key.strip()
        if not api_key:
            raise AuthConfigError("--apikey value is empty after stripping whitespace")
        header = getattr(args, "apikey_header", None) or "X-API-Key"
        sess.headers[header] = api_key
        logger.debug("API key header applied: %s", header)

# ----------------------- Funtion _apply_mtls ----------------------------#
def _apply_mtls(sess: requests.Session, args) -> None:
    cert = getattr(args, "client_cert", None)
    key = getattr(args, "client_key", None)
    if cert and key:
        if not Path(cert).is_file():
            raise AuthConfigError(f"Client certificate file not found: {cert}")
        if not Path(key).is_file():
            raise AuthConfigError(f"Client key file not found: {key}")
        if getattr(args, "cert_password", None):
            logger.warning("Provided --cert-password is not used by requests; supply an unencrypted PEM key instead.")
        sess.cert = (cert, key)
        logger.debug("mTLS client cert configured")

# ----------------------- Funtion _format_bearer ----------------------------#
def _format_bearer(token: str) -> str:
    return token if token.lower().startswith("bearer ") else f"Bearer {token}"

# ----------------------- Funtion _oauth_client_credentials ----------------------------#
def _oauth_client_credentials(args) -> str:
    if OAuth2Session is None or BackendApplicationClient is None:
        raise AuthConfigError(
            "OAuth2 dependencies missing. Install: pip install requests-oauthlib oauthlib"
        )

    cid = getattr(args, "client_id", None)
    csec = getattr(args, "client_secret", None)
    token_url = getattr(args, "token_url", None)
    scope = getattr(args, "scope", None)

    missing = [
        n for n, v in (
            ("client_id", cid),
            ("client_secret", csec),
            ("token_url", token_url),
        ) if not v
    ]
    if missing:
        raise AuthConfigError(f"--flow client requires: {', '.join(missing)}")

    client = BackendApplicationClient(client_id=cid)
    oauth = OAuth2Session(client=client)

    scope_list = None
    if isinstance(scope, str):
        scope_list = scope.split()
    elif scope:
        scope_list = scope

    if scope_list:
        token = oauth.fetch_token(
            token_url=token_url,
            client_id=cid,
            client_secret=csec,
            scope=scope_list,
            timeout=30,
        )
    else:
        token = oauth.fetch_token(
            token_url=token_url,
            client_id=cid,
            client_secret=csec,
            timeout=30,
        )

    return token["access_token"]

# ----------------------- Funtion _oauth_authorization_code ----------------------------#
def _oauth_authorization_code(args) -> str:
    if OAuth2Session is None or WebApplicationClient is None:
        raise AuthConfigError(
            "OAuth2 dependencies missing. Install: pip install requests-oauthlib oauthlib"
        )

    cid = getattr(args, "client_id", None)
    auth_url = getattr(args, "auth_url", None)
    token_url = getattr(args, "token_url", None)
    redirect_uri = getattr(args, "redirect_uri", None)
    scope = getattr(args, "scope", None)

    missing = [
        n for n, v in (
            ("client_id", cid),
            ("auth_url", auth_url),
            ("token_url", token_url),
            ("redirect_uri", redirect_uri),
        ) if not v
    ]
    if missing:
        raise AuthConfigError(f"--flow auth requires: {', '.join(missing)}")

    parsed = urlparse(redirect_uri)
    host = parsed.hostname or "127.0.0.1"
    port = parsed.port or 8765

    server, _t = _start_callback_server(host, port)

    try:
        client = WebApplicationClient(client_id=cid)
        oauth = OAuth2Session(
            client=client,
            redirect_uri=redirect_uri,
            scope=scope.split() if isinstance(scope, str) else scope,
        )

        url, state = oauth.authorization_url(auth_url)
        logger.info("Opening browser for OAuth2 authorization: %s", url)
        webbrowser.open(url)

        authorization_response = _wait_for_callback(server)

        try:
            from urllib.parse import urlsplit, parse_qs
            qs = parse_qs(urlsplit(authorization_response).query)
            returned_state = (qs.get("state") or [None])[0]
        except Exception:
            returned_state = None

        if state and returned_state != state:
            raise AuthConfigError("OAuth2 state mismatch in redirect callback")

        token = oauth.fetch_token(
            token_url=token_url,
            authorization_response=authorization_response,
            client_secret=getattr(args, "client_secret", None),
            include_client_id=True,
            timeout=30,
        )

        return token["access_token"]

    finally:
        try:
            server.shutdown()
        except Exception:
            pass


# ----------------------- Funtion configure_authentication ----------------------------#
def configure_authentication(args) -> requests.Session:
    sess = requests.Session()
    insecure = bool(getattr(args, "insecure", False))
    sess.verify = not insecure
    if insecure:
        logger.warning("TLS verification disabled (--insecure). Use only in test labs.")

    _apply_api_key(sess, args)
    _apply_mtls(sess, args)

    # Apply custom headers (--header "Name: value")
    for raw_header in getattr(args, "header", None) or []:
        if ":" in raw_header:
            name, value = raw_header.split(":", 1)
            sess.headers[name.strip()] = value.strip()
            logger.debug("Custom header applied: %s", name.strip())
        else:
            logger.warning("Ignoring malformed --header value: %s", raw_header)

    flow = getattr(args, "flow", None) or "none"

    if flow in ("none", None) and getattr(args, "token", None):
        sess.headers["Authorization"] = _format_bearer(args.token.strip())
        logger.info("Bearer token applied (flow=none).")

    if flow == "none" or flow is None:
        return sess

    if flow == "token":
        token = getattr(args, "token", None)
        if not token:
            raise AuthConfigError("--flow token requires --token")
        sess.headers["Authorization"] = _format_bearer(token.strip())
        return sess

    if flow == "client":
        access_token = _oauth_client_credentials(args)
        sess.headers["Authorization"] = f"Bearer {access_token}"
        return sess

    if flow == "basic":
        basic = getattr(args, "basic_auth", None)
        if not basic or ":" not in basic:
            raise AuthConfigError("--flow basic requires --basic-auth user:password")
        user, pwd = basic.split(":", 1)
        sess.auth = HTTPBasicAuth(user, pwd)
        return sess

    if flow == "digest":
        basic = getattr(args, "basic_auth", None)
        if not basic or ":" not in basic:
            raise AuthConfigError("--flow digest requires --basic-auth user:password")
        user, pwd = basic.split(":", 1)
        sess.auth = HTTPDigestAuth(user, pwd)
        logger.info("HTTP Digest auth configured")
        return sess

    if flow == "ntlm":
        ntlm_str = getattr(args, "ntlm", None)
        if not ntlm_str:
            raise AuthConfigError("--flow ntlm requires --ntlm DOMAIN\\user:password (or user:password)")

        m = re.match(r"(.+)\\(.+):(.+)", ntlm_str)
        if m:
            domain, user, pwd = m.groups()
            if HttpNtlmAuth is None:
                raise AuthConfigError("requests-ntlm not installed. Install: pip install requests-ntlm")
            sess.auth = HttpNtlmAuth(f"{domain}\\\\{user}", pwd)
            return sess
        m = re.match(r"([^\\:]+):(.+)", ntlm_str)
        if m:
            user, pwd = m.groups()
            if HttpNtlmAuth is None:
                raise AuthConfigError("requests-ntlm not installed. Install: pip install requests-ntlm")
            sess.auth = HttpNtlmAuth(user, pwd)
            return sess
        raise AuthConfigError("Invalid NTLM value. Use DOMAIN\\\\user:password or user:password")

    if flow == "auth":
        access_token = _oauth_authorization_code(args)
        sess.headers["Authorization"] = f"Bearer {access_token}"
        return sess

    raise AuthConfigError(f"Unknown flow: {flow}")