support for aws credential_process

alohomora/__init__.py

@@ -31,6 +31,10 @@
 __description__ = 'Get AWS API keys for a SAML-federated identity'
 
 
+def eprint(*args, **kwargs):
+ print(*args, file=sys.stderr, **kwargs)
+
+
 def die(msg):
  """Exit with non-zero and a message"""
  print(msg)

alohomora/keys.py

@@ -16,7 +16,9 @@
 
 from __future__ import print_function
 
+import json
 import os
+from datetime import datetime, timezone
 
 try:
  import ConfigParser
@@ -25,11 +27,13 @@
 
 import boto3
 
+import alohomora
 
 # https://docs.aws.amazon.com/IAM/latest/UserGuide/troubleshoot_saml.html#troubleshoot_saml_duration-exceeds
 DURATION_MIN = 15*60
 DURATION_MAX = 12*60*60
 
+
 def get(role_arn, principal_arn, assertion, duration):
  """Use the assertion to get an AWS STS token using Assume Role with SAML"""
  # We must use a session with a govcloud region for govcloud accounts
@@ -48,6 +52,7 @@ def get(role_arn, principal_arn, assertion, duration):
 
 def save(token, profile):
  """Write the AWS STS token into the AWS credential file"""
+
  filename = os.path.expanduser("~/.aws/credentials")
 
  # Read in the existing config file
@@ -76,7 +81,82 @@ def save(token, profile):
  config.write(configfile)
 
  # Give the user some basic info as to what has just happened
- print("""\n\n----------------------------------------------------------------
+ alohomora.eprint("""\n\n----------------------------------------------------------------
 Your new access key pair has been stored in the AWS configuration file {0} under the {1} profile.'
 To use this credential, call the AWS CLI with the --profile option (e.g. aws --profile {1} ec2 describe-instances).'
 ----------------------------------------------------------------\n\n""".format(filename, profile))
+
+
+def print_token(token):
+ """Print the AWS STS token to STDOUT for use with aws cli credential_process."""
+ print(json.dumps({
+ "Version": 1,
+ "AccessKeyId": token['Credentials']['AccessKeyId'],
+ "SecretAccessKey": token['Credentials']['SecretAccessKey'],
+ "SessionToken": token['Credentials']['SessionToken'],
+ "Expiration": token['Credentials']['Expiration'].isoformat()
+ }, indent=4))
+
+
+def cache(token, alohomora_profile):
+ """Write the AWS STS token to a cache file with Expiration Time."""
+
+ filename = os.path.expanduser("~/.alohomora.cache")
+ # Read in the existing cache file
+ cache = ConfigParser.RawConfigParser()
+ try:
+ cache.read(filename)
+ except ConfigParser.Error:
+ alohomora.eprint('Error reading your ~/.alohomora.cache configuration file.')
+
+ # This makes sure there is a [default] section if that's where
+ # the caller wants to put the profile. Don't ask me why this
+ # works when the cache.has_section() test below doesn't. Config
+ # be strange.
+ #
+ if alohomora_profile.lower() == 'default' and 'default' not in cache.sections():
+ cache.add_section('default')
+
+ # Put the credentials into a saml specific section instead of clobbering
+ # the default credentials
+ if not cache.has_section(alohomora_profile):
+ cache.add_section(alohomora_profile)
+
+ cache.set(alohomora_profile, 'aws_access_key_id', token['Credentials']['AccessKeyId'])
+ cache.set(alohomora_profile, 'aws_secret_access_key', token['Credentials']['SecretAccessKey'])
+ cache.set(alohomora_profile, 'aws_session_token', token['Credentials']['SessionToken'])
+ cache.set(alohomora_profile, 'exiration_time', token['Credentials']['Expiration'].isoformat())
+
+ # Write the updated cache file
+ with open(filename, 'w+') as cachefile:
+ cache.write(cachefile)
+
+
+def get_cache(alohomora_profile):
+ """Read and return non-expired cached credentials."""
+ filename = os.path.expanduser("~/.alohomora.cache")
+ # Read in the existing cache file
+ cache = ConfigParser.RawConfigParser()
+ try:
+ cache.read(filename)
+ except ConfigParser.Error:
+ alohomora.eprint('Error reading your ~/.alohomora.cache configuration file.')
+ return None
+
+ if not cache.has_section(alohomora_profile):
+ return None
+
+ token = {'Credentials': {}}
+ token['Credentials']['AccessKeyId'] = cache.get(alohomora_profile, 'aws_access_key_id')
+ token['Credentials']['SecretAccessKey'] = cache.get(alohomora_profile, 'aws_secret_access_key')
+ token['Credentials']['SessionToken'] = cache.get(alohomora_profile, 'aws_session_token')
+ token['Credentials']['Expiration'] = datetime.fromisoformat(cache.get(alohomora_profile, 'exiration_time'))
+
+ if token['Credentials']['Expiration'] > datetime.now(tz=timezone.utc):
+ return token
+
+ # Remove cached credentials that are now stale
+ cache.remove_section(alohomora_profile)
+ # Write the updated cache file
+ with open(filename, 'w+') as cachefile:
+ cache.write(cachefile)

alohomora/main.py

@@ -30,12 +30,11 @@
 except ImportError:
  import configparser as ConfigParser
 
-
 import alohomora
 import alohomora.keys
-from alohomora.keys import DURATION_MIN, DURATION_MAX
 import alohomora.req
 import alohomora.saml
+from alohomora.keys import DURATION_MAX, DURATION_MIN
 
 DEFAULT_AWS_PROFILE = 'saml'
 DEFAULT_ALOHOMORA_PROFILE = 'default'
@@ -129,6 +128,11 @@ def __init__(self):
  action='store_true',
  help="Print the program version and exit",
  default=False)
+ parser.add_argument("--aws-cli",
+ action='store_true',
+ help="Print output to stdout for credential_process",
+ default=False)
+
  self.options = parser.parse_args()
 
  # if debug is passed, set log level to DEBUG
@@ -143,7 +147,7 @@ def __init__(self):
  self.config = ConfigParser.RawConfigParser()
  self.config.read(filename)
  except ConfigParser.Error:
- print('Error reading your ~/.alohomora configuration file.')
+ alohomora.eprint('Error reading your ~/.alohomora configuration file.')
  raise
 
  def main(self):
@@ -164,81 +168,90 @@ def main(self):
  DURATION_MIN,
  DURATION_MAX))
 
- #
- # Get the user's credentials
- #
- username = self._get_config('username', os.getenv("USER"))
- if not username:
- alohomora.die("Oops, don't forget to provide a username")
+ # check for cached credentials if using credential_process output
+ awscli = self._get_config('aws-cli', False)
+ token = None
+ if awscli:
+ token = alohomora.keys.get_cache(alohomora_profile=self.__get_alohomora_profile_name())
+ if not token:
+ #
+ # Get the user's credentials
+ #
+ username = self._get_config('username', os.getenv("USER"))
+ if not username:
+ alohomora.die("Oops, don't forget to provide a username")
+
+ password = getpass.getpass(stream=sys.stderr)
+
+ idp_url = self._get_config('idp-url', None)
+ if not idp_url:
+ alohomora.die("Oops, don't forget to provide an idp-url")
+
+ auth_method = self._get_config('auth-method', None)
+ auth_device = self._get_config('auth-device', None)
+
+ #
+ # Authenticate the user
+ #
+ provider = alohomora.req.DuoRequestsProvider(idp_url, auth_method)
+ (okay, response) = provider.login_one_factor(username, password)
+ assertion = None
 
- password = getpass.getpass()
-
- idp_url = self._get_config('idp-url', None)
- if not idp_url:
- alohomora.die("Oops, don't forget to provide an idp-url")
-
- auth_method = self._get_config('auth-method', None)
- auth_device = self._get_config('auth-device', None)
-
- #
- # Authenticate the user
- #
- provider = alohomora.req.DuoRequestsProvider(idp_url, auth_method)
- (okay, response) = provider.login_one_factor(username, password)
- assertion = None
-
- if not okay:
- # we need to 2FA
- LOG.info('We need to two-factor')
- (okay, response) = provider.login_two_factor(response, auth_device)
  if not okay:
- alohomora.die('Error doing two-factor, sorry.')
- assertion = response
- else:
- LOG.info('One-factor OK')
- assertion = response
-
- awsroles = alohomora.saml.get_roles(assertion)
-
- # If I have more than one role, ask the user which one they want,
- # otherwise just proceed
- if len(awsroles) == 0:
- print('You are not authorized for any AWS roles.')
- sys.exit(0)
- elif len(awsroles) == 1:
- role_arn = awsroles[0].split(',')[0]
- principal_arn = awsroles[0].split(',')[1]
- elif len(awsroles) > 1:
- # arn:{{ partition }}:iam::{{ accountid }}:role/{{ role_name }}
- account_id = self._get_config('account', None)
- role_name = self._get_config('role-name', None)
- idp_name = self._get_config('idp-name', DEFAULT_IDP_NAME)
-
- # If the user has specified a partition, use it; otherwise, try autodiscovery
- partition = self._get_config('aws-partition', None)
- if partition is None:
- partition = awsroles[0].split(':')[1]
-
- if account_id is not None and role_name is not None and idp_name is not None:
- role_arn = "arn:%s:iam::%s:role/%s" % (partition, account_id, role_name)
- principal_arn = "arn:%s:iam::%s:saml-provider/%s" % (partition, account_id, idp_name)
+ # we need to 2FA
+ LOG.info('We need to two-factor')
+ (okay, response) = provider.login_two_factor(response, auth_device)
+ if not okay:
+ alohomora.die('Error doing two-factor, sorry.')
+ assertion = response
  else:
- account_map = {}
- try:
- accounts = self.config.options('account_map')
- for account in accounts:
- account_map[account] = self.config.get('account_map', account)
- except Exception:
- pass
- selectedrole = alohomora._prompt_for_a_thing(
- "Please choose the role you would like to assume:",
- awsroles,
- lambda s: format_role(s.split(',')[0], account_map))
-
- role_arn = selectedrole.split(',')[0]
- principal_arn = selectedrole.split(',')[1]
-
- token = alohomora.keys.get(role_arn, principal_arn, assertion, duration)
+ LOG.info('One-factor OK')
+ assertion = response
+
+ awsroles = alohomora.saml.get_roles(assertion)
+
+ # If I have more than one role, ask the user which one they want,
+ # otherwise just proceed
+ if len(awsroles) == 0:
+ alohomora.eprint('You are not authorized for any AWS roles.')
+ sys.exit(0)
+ elif len(awsroles) == 1:
+ role_arn = awsroles[0].split(',')[0]
+ principal_arn = awsroles[0].split(',')[1]
+ elif len(awsroles) > 1:
+ # arn:{{ partition }}:iam::{{ accountid }}:role/{{ role_name }}
+ account_id = self._get_config('account', None)
+ role_name = self._get_config('role-name', None)
+ idp_name = self._get_config('idp-name', DEFAULT_IDP_NAME)
+
+ # If the user has specified a partition, use it; otherwise, try autodiscovery
+ partition = self._get_config('aws-partition', None)
+ if partition is None:
+ partition = awsroles[0].split(':')[1]
+
+ if account_id is not None and role_name is not None and idp_name is not None:
+ role_arn = "arn:%s:iam::%s:role/%s" % (partition, account_id, role_name)
+ principal_arn = "arn:%s:iam::%s:saml-provider/%s" % (partition, account_id, idp_name)
+ else:
+ account_map = {}
+ try:
+ accounts = self.config.options('account_map')
+ for account in accounts:
+ account_map[account] = self.config.get('account_map', account)
+ except Exception:
+ pass
+ selectedrole = alohomora._prompt_for_a_thing(
+ "Please choose the role you would like to assume:",
+ awsroles,
+ lambda s: format_role(s.split(',')[0], account_map))
+
+ role_arn = selectedrole.split(',')[0]
+ principal_arn = selectedrole.split(',')[1]
+
+ token = alohomora.keys.get(role_arn, principal_arn, assertion, duration)
+ if awscli:
+ alohomora.keys.print_token(token)
+ alohomora.keys.cache(token, alohomora_profile=self.__get_alohomora_profile_name())
  alohomora.keys.save(token, profile=self._get_config('aws-profile', DEFAULT_AWS_PROFILE))
 
  def __get_alohomora_profile_name(self):
@@ -261,7 +274,8 @@ def _get_config(self, name, default):
  except ConfigParser.NoOptionError:
  pass
  except ConfigParser.Error:
- print('Error reading your ~/.alohomora configuration file. The file is either missing or improperly formatted.')
+ alohomora.eprint(
+ 'Error reading your ~/.alohomora configuration file. The file is either missing or improperly formatted.')
  raise
 
  data = default