Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Note: This docstring is also used by this script's command line help.
2"""A one-stop helper for desktop app to acquire an authorization code.
4It starts a web server to listen redirect_uri, waiting for auth code.
5It optionally opens a browser window to guide a human user to manually login.
6After obtaining an auth code, the web server will automatically shut down.
7"""
8from collections import defaultdict
9import logging
10import os
11import socket
12import sys
13from string import Template
14import threading
15import time
17try: # Python 3
18 from http.server import HTTPServer, BaseHTTPRequestHandler
19 from urllib.parse import urlparse, parse_qs, urlencode
20 from html import escape
21except ImportError: # Fall back to Python 2
22 from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
23 from urlparse import urlparse, parse_qs
24 from urllib import urlencode
25 from cgi import escape
28logger = logging.getLogger(__name__)
31def obtain_auth_code(listen_port, auth_uri=None): # Historically only used in testing
32 with AuthCodeReceiver(port=listen_port) as receiver:
33 return receiver.get_auth_response(
34 auth_uri=auth_uri,
35 welcome_template="""<html><body>
36 Open this link to <a href='$auth_uri'>Sign In</a>
37 (You may want to use incognito window)
38 <hr><a href='$abort_uri'>Abort</a>
39 </body></html>""",
40 ).get("code")
43def _is_inside_docker():
44 # Marker files created by the container runtimes themselves are the most
45 # reliable signal. "/.dockerenv" is created by Docker (including Docker
46 # Desktop on Mac); "/run/.containerenv" is created by Podman.
47 if os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv"):
48 return True
49 try:
50 with open("/proc/1/cgroup") as f: # https://stackoverflow.com/a/20012536/728675
51 # Search keyword "/proc/pid/cgroup" in this link for the file format
52 # https://man7.org/linux/man-pages/man7/cgroups.7.html
53 for line in f.readlines():
54 cgroup_path = line.split(":", 2)[2].strip()
55 # Only a recognized container cgroup counts as "inside a container".
56 # We must NOT treat any non-"/" path as a container: on a cgroups v2
57 # host, systemd puts PID 1 in "/init.scope" (rather than "/"), which
58 # is not a container and would otherwise cause this server to bind to
59 # 0.0.0.0 (all interfaces) instead of loopback. See issue #886.
60 if any(marker in cgroup_path for marker in (
61 "docker", "containerd", "kubepods", "lxc", "libpod")):
62 return True
63 except IOError:
64 pass # We are probably not running on Linux
65 return False
68def is_wsl():
69 # "Official" way of detecting WSL: https://github.com/Microsoft/WSL/issues/423#issuecomment-221627364
70 # Run `uname -a` to get 'release' without python
71 # - WSL 1: '4.4.0-19041-Microsoft'
72 # - WSL 2: '4.19.128-microsoft-standard'
73 import platform
74 uname = platform.uname()
75 platform_name = getattr(uname, 'system', uname[0]).lower()
76 release = getattr(uname, 'release', uname[2]).lower()
77 return platform_name == 'linux' and 'microsoft' in release
80def _browse(auth_uri, browser_name=None): # throws ImportError, webbrowser.Error
81 """Browse uri with named browser. Default browser is customizable by $BROWSER"""
82 try:
83 parsed_uri = urlparse(auth_uri)
84 if parsed_uri.scheme not in ("http", "https"):
85 logger.warning("Invalid URI scheme for browser: %s", parsed_uri.scheme)
86 return False
87 except ValueError:
88 logger.warning("Invalid URI: %s", auth_uri)
89 return False
90 if any(c in auth_uri for c in "\n\r\t"):
91 logger.warning("Invalid characters in URI")
92 return False
94 import webbrowser # Lazy import. Some distro may not have this.
95 if browser_name:
96 browser_opened = webbrowser.get(browser_name).open(auth_uri)
97 else:
98 # This one can survive BROWSER=nonexist, while get(None).open(...) can not
99 browser_opened = webbrowser.open(auth_uri)
101 # In WSL which doesn't have www-browser, try launching browser with explorer.exe
102 if not browser_opened and is_wsl():
103 import subprocess
104 try: # Try wslview first, which is the recommended way on WSL
105 # https://github.com/wslutilities/wslu
106 exit_code = subprocess.call(['wslview', auth_uri])
107 browser_opened = exit_code == 0
108 except FileNotFoundError: # wslview might not be installed
109 pass
110 if not browser_opened:
111 try:
112 # Fallback to explorer.exe as recommended for WSL
113 # Note: explorer.exe returns 1 on success in some WSL environments
114 exit_code = subprocess.call(['explorer.exe', auth_uri])
115 browser_opened = exit_code in (0, 1)
116 except FileNotFoundError:
117 pass
118 return browser_opened
121def _qs2kv(qs):
122 """Flatten parse_qs()'s single-item lists into the item itself"""
123 return {k: v[0] if isinstance(v, list) and len(v) == 1 else v
124 for k, v in qs.items()}
127def _is_html(text):
128 return text.startswith("<") # Good enough for our purpose
131def _escape(key_value_pairs):
132 return {k: escape(v) for k, v in key_value_pairs.items()}
134def _printify(text):
135 # If an https request is sent to an http server, the text needs to be repr-ed
136 return repr(text) if isinstance(text, str) and not text.isprintable() else text
138class _AuthCodeHandler(BaseHTTPRequestHandler):
139 def do_GET(self):
140 qs = parse_qs(urlparse(self.path).query)
141 welcome_param = qs.get('welcome', [None])[0]
142 error_param = qs.get('error', [None])[0]
143 if welcome_param == 'true': # Useful in manual e2e tests
144 self._send_full_response(self.server.welcome_page)
145 elif error_param == 'abort': # Useful in manual e2e tests
146 self._send_full_response("Authentication aborted", is_ok=False)
147 elif qs:
148 # GET request with auth code or error - reject for security (form_post only)
149 self._send_full_response(
150 "response_mode=query is not supported for authentication responses. "
151 "This application operates in response_mode=form_post mode only.",
152 is_ok=False)
153 else:
154 # IdP may have error scenarios that result in a parameter-less GET request
155 self._send_full_response(
156 "Authentication could not be completed. You can close this window and return to the application.",
157 is_ok=False)
158 # NOTE: Don't do self.server.shutdown() here. It'll halt the server.
160 def do_POST(self): # Handle form_post response where auth code is in body
161 # For flexibility, we choose to not check self.path matching redirect_uri
162 #assert self.path.startswith('/THE_PATH_REGISTERED_BY_THE_APP')
163 content_length = int(self.headers.get('Content-Length', 0))
164 post_data = self.rfile.read(content_length).decode('utf-8')
165 qs = parse_qs(post_data)
166 if qs.get('code') or qs.get('error'): # So, it is an auth response
167 self._process_auth_response(_qs2kv(qs))
168 else:
169 self._send_full_response("Invalid POST request", is_ok=False)
170 # NOTE: Don't do self.server.shutdown() here. It'll halt the server.
172 def _process_auth_response(self, auth_response):
173 """Process the auth response from either GET or POST request."""
174 logger.debug("Got auth response: %s", auth_response)
175 if self.server.auth_state and self.server.auth_state != auth_response.get("state"):
176 # OAuth2 successful and error responses contain state when it was used
177 # https://www.rfc-editor.org/rfc/rfc6749#section-4.2.2.1
178 self._send_full_response( # Possibly an attack
179 "State mismatch. Waiting for next response... or you may abort.", is_ok=False)
180 else:
181 template = (self.server.success_template
182 if "code" in auth_response else self.server.error_template)
183 if _is_html(template.template):
184 safe_data = _escape(auth_response) # Foiling an XSS attack
185 else:
186 safe_data = auth_response
187 filled_data = defaultdict(str, safe_data) # So that missing keys will be empty string
188 self._send_full_response(template.safe_substitute(**filled_data))
189 self.server.auth_response = auth_response # Set it now, after the response is likely sent
191 def _send_full_response(self, body, is_ok=True):
192 self.send_response(200 if is_ok else 400)
193 content_type = 'text/html' if _is_html(body) else 'text/plain'
194 self.send_header('Content-type', content_type)
195 self.end_headers()
196 self.wfile.write(body.encode("utf-8"))
198 def log_message(self, format, *args):
199 # To override the default log-to-stderr behavior
200 logger.debug(format, *map(_printify, args))
203class _AuthCodeHttpServer(HTTPServer, object):
204 def __init__(self, server_address, *args, **kwargs):
205 _, port = server_address
206 if port and (sys.platform == "win32" or is_wsl()):
207 # The default allow_reuse_address is True. It works fine on non-Windows.
208 # On Windows, it undesirably allows multiple servers listening on same port,
209 # yet the second server would not receive any incoming request.
210 # So, we need to turn it off.
211 self.allow_reuse_address = False
212 super(_AuthCodeHttpServer, self).__init__(server_address, *args, **kwargs)
214 def handle_timeout(self):
215 # It will be triggered when no request comes in self.timeout seconds.
216 # See https://docs.python.org/3/library/socketserver.html#socketserver.BaseServer.handle_timeout
217 raise RuntimeError("Timeout. No auth response arrived.") # Terminates this server
218 # We choose to not call self.server_close() here,
219 # because it would cause a socket.error exception in handle_request(),
220 # and likely end up the server being server_close() twice.
223class _AuthCodeHttpServer6(_AuthCodeHttpServer):
224 address_family = socket.AF_INET6
227class AuthCodeReceiver(object):
228 # This class has (rather than is) an _AuthCodeHttpServer, so it does not leak API
229 def __init__(self, port=None, scheduled_actions=None):
230 """Create a Receiver waiting for incoming auth response.
232 :param port:
233 The local web server will listen at http://...:<port>
234 You need to use the same port when you register with your app.
235 If your Identity Provider supports dynamic port, you can use port=0 here.
236 Port 0 means to use an arbitrary unused port, per this official example:
237 https://docs.python.org/2.7/library/socketserver.html#asynchronous-mixins
239 :param scheduled_actions:
240 For example, if the input is
241 ``[(10, lambda: print("Got stuck during sign in? Call 800-000-0000"))]``
242 then the receiver would call that lambda function after
243 waiting the response for 10 seconds.
244 """
245 address = "0.0.0.0" if _is_inside_docker() else "127.0.0.1" # Hardcode
246 # Per RFC 8252 (https://tools.ietf.org/html/rfc8252#section-8.3):
247 # * Clients should listen on the loopback network interface only.
248 # (It is not recommended to use "" shortcut to bind all addr.)
249 # * the use of localhost is NOT RECOMMENDED.
250 # (Use) the loopback IP literal
251 # rather than localhost avoids inadvertently listening on network
252 # interfaces other than the loopback interface.
253 # Note:
254 # When this server physically listens to a specific IP (as it should),
255 # you will still be able to specify your redirect_uri using either
256 # IP (e.g. 127.0.0.1) or localhost, whichever matches your registration.
257 self._scheduled_actions = sorted(scheduled_actions or []) # Make a copy
258 Server = _AuthCodeHttpServer6 if ":" in address else _AuthCodeHttpServer
259 # TODO: But, it would treat "localhost" or "" as IPv4.
260 # If pressed, we might just expose a family parameter to caller.
261 self._server = Server((address, port or 0), _AuthCodeHandler)
262 self._closing = False
264 def get_port(self):
265 """The port this server actually listening to"""
266 # https://docs.python.org/2.7/library/socketserver.html#SocketServer.BaseServer.server_address
267 return self._server.server_address[1]
269 def get_auth_response(self, timeout=None, **kwargs):
270 """Wait and return the auth response. Raise RuntimeError when timeout.
272 :param str auth_uri:
273 If provided, this function will try to open a local browser.
274 Starting from 2026, the built-in http server will require response_mode=form_post.
275 :param int timeout: In seconds. None means wait indefinitely.
276 :param str state:
277 You may provide the state you used in auth_uri,
278 then we will use it to validate incoming response.
279 :param str welcome_template:
280 If provided, your end user will see it instead of the auth_uri.
281 When present, it shall be a plaintext or html template following
282 `Python Template string syntax <https://docs.python.org/3/library/string.html#template-strings>`_,
283 and include some of these placeholders: $auth_uri and $abort_uri.
284 :param str success_template:
285 The page will be displayed when authentication was largely successful.
286 Placeholders can be any of these:
287 https://tools.ietf.org/html/rfc6749#section-5.1
288 :param str error_template:
289 The page will be displayed when authentication encountered error.
290 Placeholders can be any of these:
291 https://tools.ietf.org/html/rfc6749#section-5.2
292 :param callable auth_uri_callback:
293 A function with the shape of lambda auth_uri: ...
294 When a browser was unable to be launch, this function will be called,
295 so that the app could tell user to manually visit the auth_uri.
296 :param str browser_name:
297 If you did
298 ``webbrowser.register("xyz", None, BackgroundBrowser("/path/to/browser"))``
299 beforehand, you can pass in the name "xyz" to use that browser.
300 The default value ``None`` means using default browser,
301 which is customizable by env var $BROWSER.
302 :return:
303 The auth response of the first leg of Auth Code flow,
304 typically {"code": "...", "state": "..."} or {"error": "...", ...}
305 See https://tools.ietf.org/html/rfc6749#section-4.1.2
306 and https://openid.net/specs/openid-connect-core-1_0.html#AuthResponse
307 Returns None when the state was mismatched, or when timeout occurred.
308 """
309 # Historically, the _get_auth_response() uses HTTPServer.handle_request(),
310 # because its handle-and-retry logic is conceptually as easy as a while loop.
311 # Also, handle_request() honors server.timeout setting, and CTRL+C simply works.
312 # All those are true when running on Linux.
313 #
314 # However, the behaviors on Windows turns out to be different.
315 # A socket server waiting for request would freeze the current thread.
316 # Neither timeout nor CTRL+C would work. End user would have to do CTRL+BREAK.
317 # https://stackoverflow.com/questions/1364173/stopping-python-using-ctrlc
318 #
319 # The solution would need to somehow put the http server into its own thread.
320 # This could be done by the pattern of ``http.server.test()`` which internally
321 # use ``ThreadingHTTPServer.serve_forever()`` (only available in Python 3.7).
322 # Or create our own thread to wrap the HTTPServer.handle_request() inside.
323 result = {} # A mutable object to be filled with thread's return value
324 t = threading.Thread(
325 target=self._get_auth_response, args=(result,), kwargs=kwargs)
326 t.daemon = True # So that it won't prevent the main thread from exiting
327 t.start()
328 begin = time.time()
329 while (time.time() - begin < timeout) if timeout else True:
330 time.sleep(1) # Short detection interval to make happy path responsive
331 if not t.is_alive(): # Then the thread has finished its job and exited
332 break
333 while (self._scheduled_actions
334 and time.time() - begin > self._scheduled_actions[0][0]):
335 _, callback = self._scheduled_actions.pop(0)
336 callback()
337 return result or None
339 def _get_auth_response(self, result, auth_uri=None, timeout=None, state=None,
340 welcome_template=None, success_template=None, error_template=None,
341 auth_uri_callback=None,
342 browser_name=None,
343 ):
344 netloc = "http://localhost:{p}".format(p=self.get_port())
345 abort_uri = "{loc}?error=abort".format(loc=netloc)
346 logger.debug("Abort by visit %s", abort_uri)
348 if auth_uri:
349 # Note to maintainers:
350 # Do not enforce response_mode=form_post by secretly hardcoding it here.
351 # Just validate it here, so we won't surprise caller by changing their auth_uri behind the scene.
352 params = parse_qs(urlparse(auth_uri).query)
353 assert params.get('response_mode', [None])[0] == 'form_post', (
354 "The built-in http server supports HTTP POST only. "
355 "The auth_uri must be built with response_mode=form_post")
357 self._server.welcome_page = Template(welcome_template or "").safe_substitute(
358 auth_uri=auth_uri, abort_uri=abort_uri)
359 if auth_uri: # Now attempt to open a local browser to visit it
360 _uri = (netloc + "?welcome=true") if welcome_template else auth_uri
361 logger.info("Open a browser on this device to visit: %s" % _uri)
362 browser_opened = False
363 try:
364 browser_opened = _browse(_uri, browser_name=browser_name)
365 except: # Had to use broad except, because the potential
366 # webbrowser.Error is purposely undefined outside of _browse().
367 # Absorb and proceed. Because browser could be manually run elsewhere.
368 logger.exception("_browse(...) unsuccessful")
369 if not browser_opened:
370 if not auth_uri_callback:
371 logger.warning(
372 "Found no browser in current environment. "
373 "If this program is being run inside a container "
374 "which either (1) has access to host network "
375 "(i.e. started by `docker run --net=host -it ...`), "
376 "or (2) published port {port} to host network "
377 "(i.e. started by `docker run -p 127.0.0.1:{port}:{port} -it ...`), "
378 "you can use browser on host to visit the following link. "
379 "Otherwise, this auth attempt would either timeout "
380 "(current timeout setting is {timeout}) "
381 "or be aborted by CTRL+C. Auth URI: {auth_uri}".format(
382 auth_uri=_uri, timeout=timeout, port=self.get_port()))
383 else: # Then it is the auth_uri_callback()'s job to inform the user
384 auth_uri_callback(_uri)
386 recommendation = "For your security: Do not share the contents of this page, the address bar, or take screenshots." # From MSRC
387 self._server.success_template = Template(success_template or
388 "Authentication complete. You can return to the application. Please close this browser tab.\n\n" + recommendation)
389 self._server.error_template = Template(error_template or
390 # Do NOT invent new placeholders in this template. Just use standard keys defined in OAuth2 RFC.
391 # Otherwise there is no obvious canonical way for caller to know what placeholders are supported.
392 # Besides, we have been using these standard keys for years. Changing now would break backward compatibility.
393 "Authentication failed. $error: $error_description. ($error_uri).\n\n" + recommendation)
395 self._server.timeout = timeout # Otherwise its handle_timeout() won't work
396 self._server.auth_response = {} # Shared with _AuthCodeHandler
397 self._server.auth_state = state # So handler will check it before sending response
398 while not self._closing: # Otherwise, the handle_request() attempt
399 # would yield noisy ValueError trace
400 # Derived from
401 # https://docs.python.org/2/library/basehttpserver.html#more-examples
402 self._server.handle_request()
403 if self._server.auth_response:
404 break
405 result.update(self._server.auth_response) # Return via writable result param
407 def close(self):
408 """Either call this eventually; or use the entire class as context manager"""
409 self._closing = True
410 self._server.server_close()
412 def __enter__(self):
413 return self
415 def __exit__(self, exc_type, exc_val, exc_tb):
416 self.close()
418# Note: Manually use or test this module by:
419# python -m path.to.this.file -h
420if __name__ == '__main__':
421 import argparse, json
422 from .oauth2 import Client
423 logging.basicConfig(level=logging.INFO)
424 p = parser = argparse.ArgumentParser(
425 formatter_class=argparse.ArgumentDefaultsHelpFormatter,
426 description=__doc__ + "The auth code received will be shown at stdout.")
427 p.add_argument(
428 '--endpoint', help="The auth endpoint for your app.",
429 default="https://login.microsoftonline.com/common/oauth2/v2.0/authorize")
430 p.add_argument('client_id', help="The client_id of your application")
431 p.add_argument('--port', type=int, default=0, help="The port in redirect_uri")
432 p.add_argument('--timeout', type=int, default=60, help="Timeout value, in second")
433 p.add_argument('--host', default="127.0.0.1", help="The host of redirect_uri")
434 p.add_argument('--scope', default=None, help="The scope list")
435 args = parser.parse_args()
436 client = Client({"authorization_endpoint": args.endpoint}, args.client_id)
437 with AuthCodeReceiver(port=args.port) as receiver:
438 flow = client.initiate_auth_code_flow(
439 scope=args.scope.split() if args.scope else None,
440 redirect_uri="http://{h}:{p}".format(h=args.host, p=receiver.get_port()),
441 )
442 print(json.dumps(receiver.get_auth_response(
443 auth_uri=flow["auth_uri"],
444 welcome_template=
445 "<a href='$auth_uri'>Sign In</a>, or <a href='$abort_uri'>Abort</a>",
446 error_template="<html>Oh no. $error</html>",
447 success_template="Oh yeah. Got $code",
448 timeout=args.timeout,
449 state=flow["state"], # Optional
450 ), indent=4))