verify_dataone_token

#!/usr/bin/env python
'''
MOVED TO: https://github.com/DataONEorg/DataONE_Operations/blob/master/scripts/d1verifytoken

This copy no longer updated.

Verify a DataONE bearer token
'''

# Stdlib
import logging
import socket
import ssl
import pprint
import argparse

# 3rd party
import cryptography.hazmat.backends
import cryptography.x509
import jwt

DATAONE_ROOT = 'cn.dataone.org'


def getDecodedCert(host=DATAONE_ROOT, ssl_port=443):
  logging.info("Retrieving certificate from %s...", host)
  sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  sock.settimeout( 10.0 )
  ssl_socket = ssl.SSLSocket(sock)
  ssl_socket.connect((host, ssl_port))
  cert = ssl_socket.getpeercert(binary_form=True)
  return cryptography.x509.load_der_x509_certificate(
    cert,
    cryptography.hazmat.backends.default_backend(),
    )


def decodeVerifyToken(token, certificate):
  '''
  Decode and verify token returning contents
  
  Args:
    token: DataONE bearer token 
    certificate: Certificate from server that signed token

  Returns:
    structure represented by token
  '''
  logging.info("Decoding and verifying token...")
  try:
    return jwt.decode(token,
                      key=certificate.public_key(),
                      algorithms=['RS256'] )
  except jwt.InvalidTokenError as e:
    logging.error("Token failed to validate. error=%s", e.message)
  return None



if __name__ == "__main__":
  parser=argparse.ArgumentParser(description=__doc__,
    formatter_class=argparse.RawDescriptionHelpFormatter)
  parser.add_argument('token', nargs='?', default=None,
                      help="Bearer token to examine")
  parser.add_argument('-l', '--log_level',
                        action='count',
                        default=1,
                        help='Set logging level, multiples for more detailed.')
  parser.add_argument('-H','--host', default=DATAONE_ROOT,
                      help='Server providing certificate ({0})'.format(DATAONE_ROOT))
  args = parser.parse_args()
  levels = [logging.WARNING, logging.INFO, logging.DEBUG]
  level = levels[min(len(levels) - 1, args.log_level)]
  logging.basicConfig(level=level,
                      format="%(asctime)s %(levelname)s %(message)s")
  certificate = getDecodedCert(args.host)
  if not args.token is None:
    result = decodeVerifyToken(args.token, certificate)
    pprint.pprint(result, indent=4)
  else:
    print("No token provided.")