""" Low level functions for generating Duo Web API calls and parsing results. """ from __future__ import absolute_import from __future__ import print_function import six __version__ = '4.5.0' import base64 import collections import datetime import email.utils import hashlib import hmac import json import os import random from time import sleep import socket import ssl import sys try: # For the optional demonstration CLI program. import argparse except ImportError as e: argparse_error = e argparse = None try: # Only needed if signing requests with timezones other than UTC. import pytz except ImportError as e: pytz = None pytz_error = e from .https_wrapper import CertValidatingHTTPSConnection DEFAULT_CA_CERTS = os.path.join(os.path.dirname(__file__), 'ca_certs.pem') def canon_params(params): """ Return a canonical string version of the given request parameters. """ # this is normalized the same as for OAuth 1.0, # http://tools.ietf.org/html/rfc5849#section-3.4.1.3.2 args = [] for (key, vals) in sorted( (six.moves.urllib.parse.quote(key, '~'), vals) for (key, vals) in list(params.items())): for val in sorted(six.moves.urllib.parse.quote(val, '~') for val in vals): args.append('%s=%s' % (key, val)) return '&'.join(args) def canonicalize(method, host, uri, params, date, sig_version, body=None): """ Return a canonical string version of the given request attributes. * method: string HTTP method * host: string hostname * uri: string uri path * params: string containing request params * date: date string for request * sig_version: signature version integer * body: request body, must be string for sig_version 4 """ if sig_version == 1: canon = [ method.upper(), host.lower(), uri, canon_params(params), ] elif sig_version == 2: canon = [ date, method.upper(), host.lower(), uri, canon_params(params), ] elif sig_version == 4: # sig_version 4 is json only canon = [ date, method.upper(), host.lower(), uri, canon_params(params), hashlib.sha512(body.encode('utf-8')).hexdigest(), ] else: raise ValueError("Unknown signature version: {}".format(sig_version)) return '\n'.join(canon) def sign(ikey, skey, method, host, uri, date, sig_version, params, body=None, digestmod=hashlib.sha512): """ Return basic authorization header line with a Duo Web API signature. """ canonical = canonicalize(method, host, uri, params, date, sig_version, body=body) if isinstance(skey, six.text_type): skey = skey.encode('utf-8') if isinstance(canonical, six.text_type): canonical = canonical.encode('utf-8') sig = hmac.new(skey, canonical, digestmod) auth = '%s:%s' % (ikey, sig.hexdigest()) if isinstance(auth, six.text_type): auth = auth.encode('utf-8') b64 = base64.b64encode(auth) if not isinstance(b64, six.text_type): b64 = b64.decode('utf-8') return 'Basic %s' % b64 def normalize_params(params): """ Return copy of params with strings listified and unicode strings utf-8 encoded. """ # urllib cannot handle unicode strings properly. quote() excepts, # and urlencode() replaces them with '?'. def encode(value): if isinstance(value, bool): if value: value = 'true' else: value = 'false' elif isinstance(value, int): value = str(value) if isinstance(value, six.text_type): return value.encode("utf-8") return value def to_list(value): if value is None or isinstance(value, six.string_types): return [value] return value return dict( (encode(key), [encode(v) for v in to_list(value)]) for (key, value) in list(params.items())) class Client(object): def __init__(self, ikey, skey, host, ca_certs=DEFAULT_CA_CERTS, sig_timezone='UTC', user_agent=('Duo API Python/' + __version__), timeout=socket._GLOBAL_DEFAULT_TIMEOUT, paging_limit=100, digestmod=hashlib.sha1, # noqa: DUO130, HMAC-SHA1 still secure sig_version=2, port=None ): """ ca_certs - Path to CA pem file. """ self.ikey = ikey self.skey = skey self.host = host self.port = port self.sig_timezone = sig_timezone if ca_certs is None: ca_certs = DEFAULT_CA_CERTS self.ca_certs = ca_certs self.user_agent = user_agent self.set_proxy(host=None, proxy_type=None) self.paging_limit = paging_limit self.digestmod = digestmod self.sig_version = sig_version # Constants for handling rate limit backoff and retries self._MAX_BACKOFF_WAIT_SECS = 32 self._INITIAL_BACKOFF_WAIT_SECS = 1 self._BACKOFF_FACTOR = 2 self._RATE_LIMITED_RESP_CODE = 429 # Default timeout is a sentinel object if timeout is socket._GLOBAL_DEFAULT_TIMEOUT: self.timeout = timeout else: self.timeout = float(timeout) if sig_version == 3: raise ValueError('sig_version 3 not supported') if sig_version == 4 and digestmod != hashlib.sha512: raise ValueError('sha512 required for sig_version 4') def set_proxy(self, host, port=None, headers=None, proxy_type='CONNECT'): """ Configure proxy for API calls. Supported proxy_type values: 'CONNECT' - HTTP proxy with CONNECT. None - Disable proxy. """ if proxy_type not in ('CONNECT', None): raise NotImplementedError('proxy_type=%s' % (proxy_type,)) self.proxy_headers = headers self.proxy_host = host self.proxy_port = port self.proxy_type = proxy_type def api_call(self, method, path, params): """ Call a Duo API method. Return a (response, data) tuple. * method: HTTP request method. E.g. "GET", "POST", or "DELETE". * path: Full path of the API endpoint. E.g. "/auth/v2/ping". * params: dict mapping from parameter name to stringified value, or a dict to be converted to json. """ params_go_in_body = method in ('POST', 'PUT', 'PATCH') if self.sig_version in (1, 2): params = normalize_params(params) # v1 and v2 canonicalization don't distinguish between # params and body. There's no separate body input. body = None elif self.sig_version == 4: if params_go_in_body: body = self.canon_json(params) params = {} else: body = '' params = normalize_params(params) if self.sig_timezone == 'UTC': now = email.utils.formatdate() elif pytz is None: raise pytz_error else: d = datetime.datetime.now(pytz.timezone(self.sig_timezone)) now = d.strftime("%a, %d %b %Y %H:%M:%S %z") auth = sign(self.ikey, self.skey, method, self.host, path, now, self.sig_version, params, body=body, digestmod=self.digestmod) headers = { 'Authorization': auth, 'Date': now, } if self.user_agent: headers['User-Agent'] = self.user_agent if params_go_in_body: if self.sig_version == 4: headers['Content-type'] = 'application/json' else: headers['Content-type'] = 'application/x-www-form-urlencoded' body = six.moves.urllib.parse.urlencode(params, doseq=True) uri = path else: body = None uri = path + '?' + six.moves.urllib.parse.urlencode(params, doseq=True) encoded_headers = {} for k, v in headers.items(): if isinstance(k, six.text_type): k = k.encode('ascii') if isinstance(v, six.text_type): v = v.encode('ascii') encoded_headers[k] = v return self._make_request(method, uri, body, encoded_headers) def _connect(self): # Host and port for the HTTP(S) connection to the API server. if self.ca_certs == 'HTTP': api_port = 80 else: api_port = 443 if self.port is not None: api_port = self.port # Host and port for outer HTTP(S) connection if proxied. if self.proxy_type is None: host = self.host port = api_port elif self.proxy_type == 'CONNECT': host = self.proxy_host port = self.proxy_port else: raise NotImplementedError('proxy_type=%s' % (self.proxy_type,)) # Create outer HTTP(S) connection. if self.ca_certs == 'HTTP': conn = six.moves.http_client.HTTPConnection(host, port) elif self.ca_certs == 'DISABLE': kwargs = {} if hasattr(ssl, '_create_unverified_context'): # httplib.HTTPSConnection validates certificates by # default in Python 2.7.9+. kwargs['context'] = ssl._create_unverified_context() # noqa: DUO122, explicitly disabled for testing scenarios conn = six.moves.http_client.HTTPSConnection(host, port, **kwargs) else: conn = CertValidatingHTTPSConnection(host, port, ca_certs=self.ca_certs) # Override default socket timeout if requested. conn.timeout = self.timeout # Configure CONNECT proxy tunnel, if any. if self.proxy_type == 'CONNECT': if hasattr(conn, 'set_tunnel'): # 2.7+ conn.set_tunnel(self.host, api_port, self.proxy_headers) elif hasattr(conn, '_set_tunnel'): # 2.6.3+ # pylint: disable=E1103 conn._set_tunnel(self.host, api_port, self.proxy_headers) # pylint: enable=E1103 return conn def _make_request(self, method, uri, body, headers): if self.proxy_type == 'CONNECT': # Ensure the request uses the correct protocol and Host. if self.ca_certs == 'HTTP': api_proto = 'http' else: api_proto = 'https' uri = ''.join((api_proto, '://', self.host, uri)) conn = self._connect() # backoff on rate limited requests and retry. if a request is rate # limited after MAX_BACKOFF_WAIT_SECS, return the rate limited response wait_secs = self._INITIAL_BACKOFF_WAIT_SECS while True: response, data = self._attempt_single_request( conn, method, uri, body, headers) if (response.status != self._RATE_LIMITED_RESP_CODE or wait_secs > self._MAX_BACKOFF_WAIT_SECS): break random_offset = random.uniform(0.0, 1.0) # noqa: DUO102, non-cryptographic random use sleep(wait_secs + random_offset) wait_secs = wait_secs * self._BACKOFF_FACTOR self._disconnect(conn) return (response, data) def _attempt_single_request(self, conn, method, uri, body, headers): conn.request(method, uri, body, headers) response = conn.getresponse() data = response.read() return (response, data) def _disconnect(self, conn): conn.close() def normalize_paging_args(self, limit=None, offset=0): """ Converts paging arguments to a format the rest of the client expects. :param limit: The number of objects requested of a paginated api endpoint. If it looks falsy, it is is not changed. Default None :param offset: The offset to start retrieval. Default 0 :return: tuple after the form of (limit, offset) """ if limit: limit = '{}'.format(limit) offset = '{}'.format(offset) return (limit, offset) def json_api_call(self, method, path, params): """ Call a Duo API method which is expected to return a JSON body with a 200 status. Return the response data structure or raise RuntimeError. """ (response, data) = self.api_call(method, path, params) return self.parse_json_response(response, data) def json_paging_api_call(self, method, path, params): """ Call a Duo API method which is expected to return a JSON body with a 200 status. Return a generator that can be used to get response data or raise a RuntimeError. """ objects = [] next_offset = 0 if 'limit' not in params and self.paging_limit: params['limit'] = str(self.paging_limit) while next_offset is not None: params['offset'] = str(next_offset) (response, data) = self.api_call(method, path, params) (objects, metadata) = self.parse_json_response_and_metadata(response, data) next_offset = metadata.get('next_offset', None) for obj in objects: yield obj def json_cursor_api_call(self, method, path, params, get_records_func): """ Call a Duo API endpoint which utilizes a cursor in some responses to page through a set of data. This cursor is supplied through the optional "offset" parameter. The cursor for the next set of data is in the response metadata as "next_offset". Callers must also include a function parameter to extract the iterable of records to yield. This is slightly different than json_paging_api_call because the first request does not contain the offset parameter. :param method: The method to make the request w/ as a string. Ex: "GET", "POST", "PUT" etc. :param path: The path to make the request with as a string. :param params: The dict of parameters to send in the request. :param get_records_func: Function that can be called to extract an iterable of records from the parsed response json. :returns: Generator which will yield records from the api response(s). """ next_offset = None if 'limit' not in params and self.paging_limit: params['limit'] = str(self.paging_limit) while True: if next_offset is not None: params['offset'] = str(next_offset) (http_resp, http_resp_data) = self.api_call(method, path, params) (response, metadata) = self.parse_json_response_and_metadata( http_resp, http_resp_data, ) for record in get_records_func(response): yield record next_offset = metadata.get('next_offset', None) if next_offset is None: break def parse_json_response(self, response, data): """ Return the parsed data structure or raise RuntimeError. """ (response, metadata) = self.parse_json_response_and_metadata(response, data) return response def parse_json_response_and_metadata(self, response, data): """ Return the parsed data structure and metadata as a tuple or raise RuntimeError. """ def raise_error(msg): error = RuntimeError(msg) error.status = response.status error.reason = response.reason error.data = data raise error if not isinstance(data, six.text_type): data = data.decode('utf-8') if response.status != 200: try: data = json.loads(data) if data['stat'] == 'FAIL': if 'message_detail' in data: raise_error('Received %s %s (%s)' % ( response.status, data['message'], data['message_detail'], )) else: raise_error('Received %s %s' % ( response.status, data['message'], )) except (ValueError, KeyError, TypeError): pass raise_error('Received %s %s' % ( response.status, response.reason, )) try: data = json.loads(data) if data['stat'] != 'OK': raise_error('Received error response: %s' % data) return (data['response'], data.get('metadata', {})) except (ValueError, KeyError, TypeError): raise_error('Received bad response: %s' % data) @classmethod def canon_json(cls, params): if not isinstance(params, dict): raise ValueError('JSON request must be an object.') return json.dumps(params, sort_keys=True, separators=(',', ':')) def output_response(response, data, headers=None): """ Print response, parsed, sorted, and pretty-printed if JSON """ if not headers: headers = [] print(response.status, response.reason) for header in headers: val = response.getheader(header) if val is not None: print('%s: %s' % (header, val)) try: if not isinstance(data, six.text_type): data = data.decode('utf-8') data = json.loads(data) data = json.dumps(data, sort_keys=True, indent=4) except ValueError: pass print(data) def main(): if argparse is None: raise argparse_error parser = argparse.ArgumentParser() # named arguments parser.add_argument('--ikey', required=True, help='Duo integration key') parser.add_argument('--skey', required=True, help='Duo integration secret key') parser.add_argument('--host', required=True, help='Duo API hostname') parser.add_argument('--method', required=True, help='HTTP request method') parser.add_argument('--path', required=True, help='API endpoint path') parser.add_argument('--ca', default=DEFAULT_CA_CERTS) parser.add_argument('--sig-version', type=int, default=2) parser.add_argument('--sig-timezone', default='UTC') parser.add_argument( '--show-header', action='append', default=[], metavar='Header-Name', help='Show specified response header(s) (default: only output body).', ) parser.add_argument('--file-args', default=[]) # optional positional arguments are used for GET/POST params, name=val parser.add_argument('param', nargs='*') args = parser.parse_args() client = Client( ikey=args.ikey, skey=args.skey, host=args.host, ca_certs=args.ca, sig_version=args.sig_version, sig_timezone=args.sig_timezone, ) params = collections.defaultdict(list) for p in args.param: try: (k, v) = p.split('=', 1) except ValueError: sys.exit('Error: Positional argument %s is not ' 'in key=value format.' % (p,)) params[k].append(v) # parse which arguments are filenames file_args = args.file_args if args.file_args: file_args = file_args.split(',') for (k, v) in list(params.items()): if k in file_args: # value is a filename, replace with contents if len(v) != 1: # file arguments cannot have multiple values raise NotImplementedError (v,) = v with open(v, 'rb') as val: params[k] = base64.b64encode(val.read()) else: params[k] = v (response, data) = client.api_call(args.method, args.path, params) output_response(response, data, args.show_header) if __name__ == '__main__': main()