24.2 Sample OAuth 2.0 Application Integrated with Advanced Authentication

Sample script (oauth2_test.py)

from bottle import Bottle, request, run, redirect, SimpleTemplate, template
from urllib.parse import urlparse, urlunparse, urlencode,quote
import urllib.request
import base64
import ssl
import json

app = Bottle()

client_id = 'id-rSCzuBLQgXCATfkXZ4fsedAo8sPsWxSs'
client_secret = 'secret-91DpzWFD26RriURR7KJ1pryFx7V9QeDm'
redirect_uri = 'http://localhost:8088/'  # this app callback URI
authorization_endpoint = 'https://192.168.0.151/osp/a/TOP/auth/oauth2/grant'
attributes_endpoint = 'https://192.168.0.151/osp/a/TOP/auth/oauth2/getattributes'
state = {}

@app.get('/getattr')
def get_attributes():
    params = urlencode({
        'attributes': 'client username userRepository user_dn user_cn mail sid upn netbiosName',
        'access_token': state['access_token']
    })
    url = attributes_endpoint + '?' + params
    print('getattr url: {}\n'.format(url))
    req = urllib.request.Request(url)
    gcontext = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)  # avoid cert checking
    with urllib.request.urlopen(req, context=gcontext) as response:  # perform GET request and read response
        rsp = response.read()
    attributes = json.loads(rsp.decode('utf-8'))
    return template('attributes.html', items=attributes.items(), refresh_token=urllib.parse.quote(state['refresh_token']))


@app.get('/')
def do_get():
    code = request.query.get('code')
    if code:
        # got code from OAuth 2 authentication server
        token = get_token_code(code)
        state.update(token)
        return template('token.html', items=token.items(), refresh_token=urllib.parse.quote(token['refresh_token']))
    else:
        return template('main.html')


@app.get('/logon')
def do_logon():
    pr=list(urlparse(authorization_endpoint))
    # set query
    pr[4]=urlencode({
        'response_type': 'code',
        'client_id': client_id,
        'redirect_uri': redirect_uri
    })
    # perform redirection to OAuth 2 authentication server
    redirect(urlunparse(pr))

@app.get('/logon-implicit')
def do_logon_implicit():
    # parse authorization_endpoint URL
    pr = list(urlparse(authorization_endpoint))
    # set query
    pr[4] = urlencode({
        'response_type': 'token',
        'client_id': client_id,
    })
    # perform redirection to OAuth 2 authentication server
    redirect(urlunparse(pr))

@app.get('/logon-creds')
def do_logon_creds():
    return template('logonform.html')

@app.post('/logon-creds')
def do_logon_creds_post():
    username = request.forms.get('username')
    password = request.forms.get('password')
    token = get_token_password(username, password)
    state.update(token)
    return template('token.html', items=token.items(), refresh_token=urllib.parse.quote(token['refresh_token']))

def get_token_password(username, password):
    # prepare POST parameters - encode them to urlencoded
    data = urlencode({
        'grant_type': 'password',
        'username': username,
        'password': password
    })
    data = data.encode('ascii')  # data should be bytes
    resp_text = post_data(data, prepare_headers())
    print(resp_text)
    return json.loads(resp_text)

@app.get('/refresh')
def do_refresh():
    token = refresh_access_token(request.query.get('refresh_token'))
    state.update(token)
    return template('token.html', items=token.items(), refresh_token=state.get('refresh_token', ''))

def get_token_code(code):
    # prepare POST parameters - encode them to urlencoded
    data = urlencode({
        'grant_type': 'authorization_code',
        'code': code,
        'redirect_uri': redirect_uri
    })
    data = data.encode('ascii')  # data should be bytes
    resp_text = post_data(data, prepare_headers())
    print(resp_text)
    return json.loads(resp_text)

def refresh_access_token(refresh_token):
    print('refresh_token: {}'.format(refresh_token))
    # prepare POST parameters - encode them to urlencoded
    data = urlencode({
        'grant_type': 'refresh_token',
        'refresh_token': refresh_token,
    })
    data = data.encode('ascii')  # data should be bytes
    resp_text = post_data(data, prepare_headers())
    print(resp_text)
    return json.loads(resp_text)

def prepare_headers(use_content_type_hdr = True):
    hdrs = {
        'Authorization': 'Basic {}'.format(base64.b64encode(
            '{}:{}'.format(quote(client_id, safe=''), quote(client_secret, safe='')).encode('ascii')).decode(
            'ascii')),
    }
    if use_content_type_hdr:
        hdrs.update({'Content-type': 'application/x-www-form-urlencoded'})
    return hdrs

def post_data(data, headers):
    print('post_data\nheaders:\n{}\ndata:\n{}'.format(headers, data))
    req = urllib.request.Request(authorization_endpoint, data, headers)
    gcontext = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)  # avoid cert checking
    with urllib.request.urlopen(req, context=gcontext) as response:  # perform POST request and read response
        rsp = response.read()
    return rsp.decode('utf-8')


run(app, host='0.0.0.0', port=8088)
 

Main menu (main.html)

<!DOCTYPE html>
<html>
<head lang="en">
    <meta charset="UTF-8">
    <title></title>
    <script type="text/javascript">
         //<![CDATA[
            function getHashParam(name) {
                var hash = window.location.hash;
                if (hash) {
                    if (name = (new RegExp('[#&]' + encodeURIComponent(name) + '=([^&]*)')).exec(hash))
                    return decodeURIComponent(name[1]);
                }
            }
            function showResult() {
                if (window.location.hash) {
                    document.getElementById('result').innerHTML = '<table border="1">'+
                        '<tr><td>access_token</td><td>'+getHashParam('access_token')+'</td></tr>'+
                        '<tr><td>token_type</td><td>'+getHashParam('token_type')+'</td></tr>'+
                        '<tr><td>expires_in</td><td>'+getHashParam('expires_in')+'</td></tr>'+
                        '</table>';
                } else {
                    document.getElementById('result').innerHTML = 'Implicit granted token is not found';
                }
            }
          ]]//>
        </script>
</head>
<body onload="showResult();">
<div id="result">result</div><br/>
<br/>
Click <a href="/logon">here</a> to obtain an authentication token through Authorization Code Grant<br/>
Click <a href="/logon-implicit">here</a> to obtain an authentication token through Implicit Grant (the token will be received in hash part of THIS page)<br/>
Click <a href="/logon-creds">here</a> to obtain an authentication token through Resource Owner Password Credentials Grant<br/>
</body>
</html>

Token information (token.html)

<!DOCTYPE html>
<html>
<head lang="en">
    <meta charset="UTF-8">
    <title></title>
</head>
<body>
Token<br/>
<table border="1">
    % for k, v in items:
    <tr>
        <td>{{k}}</td>
        <td>{{v}}</td>
    </tr>
    % end
</table>
<br/>
<a href="/getattr">Get attributes</a><br/>
<a href="/refresh?refresh_token={{refresh_token}}">Refresh token</a>
</body>
</html> 

Attributes information (attributes.html)

<!DOCTYPE html>
<html>
<head lang="en">
    <meta charset="UTF-8">
    <title></title>
</head>
<body>
Attributes<br/>
<table border="1">
    % for k, v in items:
    <tr>
        <td>{{k}}</td>
        <td>{{v}}</td>
    </tr>
    % end
</table>
<br/>
<a href="/refresh?refresh_token={{refresh_token}}">Refresh token</a>
</body>
</html>

Logon form for Resource Owner Password Credentials Grant mode (logonform.html)

<!DOCTYPE html>
<html>
<head lang="en">
    <meta charset="UTF-8">
    <title></title>
</head>
<body>
<form method="post" action="/logon-creds">
    User name: <input type="text" name="username"><br/>
    Password: <input type="password" name="password"><br/>
    <input type="submit">
</form>
</body>
</html>