Merge remote-tracking branch 'oauth2cli/dev' into oauth2

Author: rayluo

oauth2cli/__init__.py

@@ -0,0 +1,4 @@
+__version__ = "0.0.1"
+
+from .oauth2 import Client
+

oauth2cli/assertion.py

@@ -0,0 +1,70 @@
+import time
+import binascii
+import base64
+import uuid
+import logging
+
+import jwt
+
+
+logger = logging.getLogger(__file__)
+
+class Signer(object):
+    def sign_assertion(
+            self, audience, issuer, subject, expires_at,
+            issued_at=None, assertion_id=None, **kwargs):
+        # Names are defined in https://tools.ietf.org/html/rfc7521#section-5
+        raise NotImplementedError("Will be implemented by sub-class")
+
+
+class JwtSigner(Signer):
+    def __init__(self, key, algorithm, sha1_thumbprint=None, headers=None):
+        """Create a signer.
+
+        Args:
+
+            key (str): The key for signing, e.g. a base64 encoded private key.
+            algorithm (str):
+                "RS256", etc.. See https://pyjwt.readthedocs.io/en/latest/algorithms.html
+                RSA and ECDSA algorithms require "pip install cryptography".
+            sha1_thumbprint (str): The x5t aka X.509 certificate SHA-1 thumbprint.
+            headers (dict): Additional headers, e.g. "kid" or "x5c" etc.
+        """
+        self.key = key
+        self.algorithm = algorithm
+        self.headers = headers or {}
+        if sha1_thumbprint:  # https://tools.ietf.org/html/rfc7515#section-4.1.7
+            self.headers["x5t"] = base64.urlsafe_b64encode(
+                binascii.a2b_hex(sha1_thumbprint)).decode()
+
+    def sign_assertion(
+            self, audience, issuer, subject=None, expires_at=None,
+            issued_at=None, assertion_id=None, not_before=None,
+            additional_claims=None, **kwargs):
+        """Sign a JWT Assertion.
+
+        Parameters are defined in https://tools.ietf.org/html/rfc7523#section-3
+        Key-value pairs in additional_claims will be added into payload as-is.
+        """
+        now = time.time()
+        payload = {
+            'aud': audience,
+            'iss': issuer,
+            'sub': subject or issuer,
+            'exp': expires_at or (now + 10*60),  # 10 minutes
+            'iat': issued_at or now,
+            'jti': assertion_id or str(uuid.uuid4()),
+            }
+        if not_before:
+            payload['nbf'] = not_before
+        payload.update(additional_claims or {})
+        try:
+            return jwt.encode(
+                payload, self.key, algorithm=self.algorithm, headers=self.headers)
+        except:
+            if self.algorithm.startswith("RS") or self.algorithm.starswith("ES"):
+                logger.exception(
+                    'Some algorithms requires "pip install cryptography". '
+                    'See https://pyjwt.readthedocs.io/en/latest/installation.html#cryptographic-dependencies-optional')
+            raise
+

oauth2cli/authcode.py

@@ -0,0 +1,111 @@
+# Note: This docstring is also used by this script's command line help.
+"""A one-stop helper for desktop app to acquire an authorization code.
+
+It starts a web server to listen redirect_uri, waiting for auth code.
+It optionally opens a browser window to guide a human user to manually login.
+After obtaining an auth code, the web server will automatically shut down.
+"""
+
+import argparse
+import webbrowser
+import logging
+
+try:  # Python 3
+    from http.server import HTTPServer, BaseHTTPRequestHandler
+    from urllib.parse import urlparse, parse_qs, urlencode
+except ImportError:  # Fall back to Python 2
+    from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
+    from urlparse import urlparse, parse_qs
+    from urllib import urlencode
+
+from .oauth2 import Client
+
+
+logger = logging.getLogger(__file__)
+
+def obtain_auth_code(listen_port, auth_uri=None):
+    """This function will start a web server listening on http://localhost:port
+    and then you need to open a browser on this device and visit your auth_uri.
+    When interaction finishes, this function will return the auth code,
+    and then shut down the local web server.
+
+    :param listen_port:
+        The local web server will listen at http://localhost:<listen_port>
+        Unless the authorization server supports dynamic port,
+        you need to use the same port when you register with your app.
+    :param auth_uri: If provided, this function will try to open a local browser.
+    :return: Hang indefinitely, until it receives and then return the auth code.
+    """
+    exit_hint = "Visit http://localhost:{p}?code=exit to abort".format(p=listen_port)
+    logger.warn(exit_hint)
+    if auth_uri:
+        page = "http://localhost:{p}?{q}".format(p=listen_port, q=urlencode({
+            "text": "Open this link to sign in. You may use incognito window",
+            "link": auth_uri,
+            "exit_hint": exit_hint,
+            }))
+        browse(page)
+    server = HTTPServer(("", int(listen_port)), AuthCodeReceiver)
+    try:
+        server.authcode = None
+        while not server.authcode:
+            # Derived from
+            # https://docs.python.org/2/library/basehttpserver.html#more-examples
+            server.handle_request()
+        return server.authcode
+    finally:
+        server.server_close()
+
+def browse(auth_uri):
+    controller = webbrowser.get()  # Get a default controller
+    # Some Linux Distro does not setup default browser properly,
+    # so we try to explicitly use some popular browser, if we found any.
+    for browser in ["chrome", "firefox", "safari", "windows-default"]:
+        try:
+            controller = webbrowser.get(browser)
+            break
+        except webbrowser.Error:
+            pass  # This browser is not installed. Try next one.
+    logger.info("Please open a browser on THIS device to visit: %s" % auth_uri)
+    controller.open(auth_uri)
+
+class AuthCodeReceiver(BaseHTTPRequestHandler):
+    def do_GET(self):
+        # For flexibility, we choose to not check self.path matching redirect_uri
+        #assert self.path.startswith('/THE_PATH_REGISTERED_BY_THE_APP')
+        qs = parse_qs(urlparse(self.path).query)
+        if qs.get('code'):  # Then store it into the server instance
+            ac = self.server.authcode = qs['code'][0]
+            self._send_full_response('Authcode:\n{}'.format(ac))
+            # NOTE: Don't do self.server.shutdown() here. It'll halt the server.
+        elif qs.get('text') and qs.get('link'):  # Then display a landing page
+            self._send_full_response(
+                '<a href={link}>{text}</a><hr/>{exit_hint}'.format(
+                link=qs['link'][0], text=qs['text'][0],
+                exit_hint=qs.get("exit_hint", [''])[0],
+                ))
+        else:
+            self._send_full_response("This web service serves your redirect_uri")
+
+    def _send_full_response(self, body, is_ok=True):
+        self.send_response(200 if is_ok else 400)
+        content_type = 'text/html' if body.startswith('<') else 'text/plain'
+        self.send_header('Content-type', content_type)
+        self.end_headers()
+        self.wfile.write(body.encode("utf-8"))
+
+
+if __name__ == '__main__':
+    logging.basicConfig(level=logging.INFO)
+    p = parser = argparse.ArgumentParser(
+        description=__doc__ + "The auth code received will be shown at stdout.")
+    p.add_argument('endpoint',
+        help="The auth endpoint for your app. For example: "
+            "https://login.microsoftonline.com/your_tenant/oauth2/authorize")
+    p.add_argument('client_id', help="The client_id of your application")
+    p.add_argument('redirect_port', type=int, help="The port in redirect_uri")
+    args = parser.parse_args()
+    client = Client(args.client_id, authorization_endpoint=args.endpoint)
+    auth_uri = client.build_auth_request_uri("code")
+    print(obtain_auth_code(args.redirect_port, auth_uri))
+