Skip to content

Instantly share code, notes, and snippets.

@ryandgoulding
Created October 28, 2015 01:11
Show Gist options
  • Save ryandgoulding/a9e07e040fae43dab2a4 to your computer and use it in GitHub Desktop.
Save ryandgoulding/a9e07e040fae43dab2a4 to your computer and use it in GitHub Desktop.
#!/usr/bin/env python
import calendar
import time
import sys
import logging
import json
import subprocess
import sqlite3
'''
get-access
@author Ryan Goulding
Usage:
kinit <user>
get-access
Gain access to ODL using default Principle
'''
class GainAccess:
"""Connector from SSSD to ODL"""
debug=False
def __init__(self):
self.log = logging.getLogger()
if self.debug:
self.log.setLevel(logging.DEBUG)
self.console = logging.StreamHandler(sys.stdout)
if self.debug:
self.console.setLevel(logging.DEBUG)
self.formatter = logging.Formatter("%(name)-12s: %(levelname)-8s %(message)s")
self.console.setFormatter(self.formatter)
self.log.addHandler(self.console)
self.conn = sqlite3.connect('keystore.db')
self.write_table();
def write_table(self):
c = self.conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS auth_tokens (
ts DATETIME DEFAULT CURRENT_TIMESTAMP,
expires_in INT NOT NULL,
token_type VARCHAR(255) NOT NULL,
access_token VARCHAR(255) NOT NULL PRIMARY KEY
);''')
c.close()
def get_auth_token(self):
sql = '''SELECT * FROM auth_tokens LIMIT 1'''
c = self.conn.cursor()
c.execute(sql)
token = c.fetchone()
ts=time.time()
if token is not None:
self.log.debug(token)
tts = calendar.timegm(time.strptime(token[0], '%Y-%m-%d %H:%M:%S'))
elapsed=ts-tts
self.log.debug("Elapsed time: %d",elapsed)
expires_in=token[1]
if elapsed > expires_in:
token = self.create_cached_token()
else:
token = self.create_cached_token()
return token
def create_cached_token(self):
self.log.debug("No access token is cached locally")
token = self.get_access_token(self.get_refresh_token())
self.log.debug(token)
sql='''DELETE FROM auth_tokens'''
c = self.conn.cursor()
c.execute(sql)
self.conn.commit()
sql='''INSERT INTO auth_tokens (expires_in,token_type,access_token) VALUES (%d,"%s","%s")''' % token
self.log.debug(sql)
c = self.conn.cursor()
c.execute(sql)
self.conn.commit()
sql = '''SELECT * FROM auth_tokens LIMIT 1'''
c = self.conn.cursor()
c.execute(sql)
return c.fetchone()
def set_debug(self):
self.debug=True
self.log.setLevel(logging.DEBUG)
self.console.setLevel(logging.DEBUG)
def execute_cli(self, cmd):
return subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE).stdout.read()
def get_refresh_token(self):
cmd="curl -s --negotiate -u : -X POST http://odl.brcd-sssd-tb.com/oauth2/federation/"
description="Requesting a refresh token"
self.report_cmd(description, cmd)
cmd_output = self.execute_cli(cmd)
self.report_output(cmd_output)
return json.loads(cmd_output)['refresh_token']
def get_access_token(self, refresh_token):
cmd="curl -s -d 'grant_type=refresh_token&refresh_token=%s&scope=sdn' http://odl.brcd-sssd-tb.com:8181/oauth2/token" % (refresh_token)
description="Get access_token"
self.report_cmd(description, cmd)
cmd_output = self.execute_cli(cmd)
self.report_output(cmd_output)
ret = json.loads(cmd_output)
return (ret['expires_in'], ret['token_type'], ret['access_token'])
def report_output(self, cmd_output):
self.log.debug("Output: %s" % (cmd_output));
def report_cmd(self, description, cmd):
self.log.debug("%s: %s" % (description, cmd))
def get_curl_statement(self, access_token):
cmd="curl -s -H 'Authorization: Bearer %s' http://odl.brcd-sssd-tb.com:8181/restconf/streams/" % (access_token)
return cmd
def do_rest_call(self, access_token):
cmd=self.get_curl_statement(access_token)
description="An example restconf call"
self.report_cmd(description, cmd)
cmd_output = self.execute_cli(cmd)
self.report_output(cmd_output)
if __name__=='__main__':
access = GainAccess();
if len(sys.argv) > 1:
access.set_debug()
print access.get_auth_token()[3]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment