Created
October 24, 2018 15:05
-
-
Save lmayorga1980/05f422b34ac272ca2ba995cc56758171 to your computer and use it in GitHub Desktop.
Python-Federated-API-SAML-ADFS
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#based on https://aws.amazon.com/blogs/security/how-to-implement-federated-api-and-cli-access-using-saml-2-0-and-ad-fs/ | |
import sys | |
import boto.sts | |
import boto.s3 | |
import requests | |
import ConfigParser | |
import base64 | |
import logging | |
import os | |
import xml.etree.ElementTree as ET | |
import re | |
from bs4 import BeautifulSoup | |
from os.path import expanduser | |
from urlparse import urlparse | |
import xml.dom.minidom | |
# region: The default AWS region that this script will connect to for all API calls | |
# Set AWS_ADFS_URL and AWS_ADFS_SESSION_TIMEOUT | |
region = "<Your aws region>" | |
username = "<Your domain user name>" | |
password = "<Your password>" | |
profilename = "<Your profile name>" | |
# output format: The AWS CLI output format that will be configured in the | |
# saml profile (affects subsequent CLI calls) | |
outputformat = 'json' | |
# awsconfigfile: The file where this script will store the temp | |
# credentials under the saml profile | |
awsconfigfile = '/.aws/credentials' | |
# SSL certificate verification: Whether or not strict certificate | |
# verification is done, False should only be used for dev/test | |
sslverification = True | |
default_creds_file = expanduser("~") + awsconfigfile | |
exists = os.path.isfile(default_creds_file) | |
if exists: | |
credentials = open(default_creds_file, "r").read() | |
if "[default]" not in credentials: | |
print "You need to have the default credentials configured even if the values are empty" | |
exit(1) | |
else: | |
print "You need to create the credentials file in your home directory" | |
exit(1) | |
# idpentryurl: The initial url that starts the authentication process. | |
#idpentryurl = 'https://<fqdn>:<port>/adfs/ls/IdpInitiatedSignOn.aspx?loginToRp=urn:amazon:webservices' | |
try: | |
idpentryurl = os.environ['AWS_ADFS_URL'] | |
except KeyError: | |
print "You need to set the AWS_ADFS_URL to a valid https URL" | |
exit(1) | |
print "Using AWS_ADFS_URL: " + idpentryurl | |
try: | |
session_timeout = os.environ['AWS_ADFS_SESSION_TIMEOUT'] | |
except KeyError: | |
print "Session Time out is not set. Using the default of 60 minutes" | |
session_timeout = 3600 | |
# Uncomment to enable low level debugging | |
logging.basicConfig(level=logging.INFO) | |
########################################################################## | |
# Initiate session handler | |
session = requests.Session() | |
# Programmatically get the SAML assertion | |
# Opens the initial IdP url and follows all of the HTTP302 redirects, and | |
# gets the resulting login page | |
formresponse = session.get(idpentryurl, verify=sslverification) | |
# Capture the idpauthformsubmiturl, which is the final url after all the 302s | |
idpauthformsubmiturl = formresponse.url | |
# Parse the response and extract all the necessary values | |
# in order to build a dictionary of all of the form values the IdP expects | |
#formsoup = BeautifulSoup(formresponse.text.decode('utf8')) | |
formsoup = BeautifulSoup(formresponse.text.decode('utf8'),"html.parser") | |
payload = {} | |
for inputtag in formsoup.find_all(re.compile('(INPUT|input)')): | |
name = inputtag.get('name','') | |
value = inputtag.get('value','') | |
if "user" in name.lower(): | |
#Make an educated guess that this is the right field for the username | |
payload[name] = username | |
elif "email" in name.lower(): | |
#Some IdPs also label the username field as 'email' | |
payload[name] = username | |
elif "pass" in name.lower(): | |
#Make an educated guess that this is the right field for the password | |
payload[name] = password | |
else: | |
#Simply populate the parameter with the existing value (picks up hidden fields in the login form) | |
payload[name] = value | |
# Debug the parameter payload if needed | |
# Use with caution since this will print sensitive output to the screen | |
#print payload | |
# Some IdPs don't explicitly set a form action, but if one is set we should | |
# build the idpauthformsubmiturl by combining the scheme and hostname | |
# from the entry url with the form action target | |
# If the action tag doesn't exist, we just stick with the | |
# idpauthformsubmiturl above | |
for inputtag in formsoup.find_all(re.compile('(FORM|form)')): | |
action = inputtag.get('action') | |
loginid = inputtag.get('id') | |
if (action and loginid == "loginForm"): | |
parsedurl = urlparse(idpentryurl) | |
idpauthformsubmiturl = parsedurl.scheme + "://" + parsedurl.netloc + action | |
# Performs the submission of the IdP login form with the above post data | |
response = session.post( | |
idpauthformsubmiturl, data=payload, verify=sslverification) | |
# Debug the response if needed | |
#print (response.text) | |
# Overwrite and delete the credential variables, just for safety | |
username = '##############################################' | |
password = '##############################################' | |
del username | |
del password | |
# Decode the response and extract the SAML assertion | |
soup = BeautifulSoup(response.text.decode('utf8'), "html.parser") | |
assertion = '' | |
# Look for the SAMLResponse attribute of the input tag (determined by | |
# analyzing the debug print lines above) | |
for inputtag in soup.find_all('input'): | |
if(inputtag.get('name') == 'SAMLResponse'): | |
#print(inputtag.get('value')) | |
assertion = inputtag.get('value') | |
# Better error handling is required for production use. | |
if (assertion == ''): | |
#TODO: Insert valid error checking/handling | |
label = soup.find(id='errorText') | |
print label.string | |
print '------------' | |
print 'Response did not contain a valid SAML assertion' | |
sys.exit(1) | |
# Debug only | |
xml = xml.dom.minidom.parseString(base64.b64decode(assertion)) # or xml.dom.minidom.parseString(xml_string) | |
print xml.toprettyxml() | |
# Parse the returned assertion and extract the authorized roles | |
awsroles = [] | |
root = ET.fromstring(base64.b64decode(assertion)) | |
for saml2attribute in root.iter('{urn:oasis:names:tc:SAML:2.0:assertion}Attribute'): | |
if (saml2attribute.get('Name') == 'https://aws.amazon.com/SAML/Attributes/Role'): | |
for saml2attributevalue in saml2attribute.iter('{urn:oasis:names:tc:SAML:2.0:assertion}AttributeValue'): | |
awsroles.append(saml2attributevalue.text) | |
# Note the format of the attribute value should be role_arn,principal_arn | |
# but lots of blogs list it as principal_arn,role_arn so let's reverse | |
# them if needed | |
for awsrole in awsroles: | |
chunks = awsrole.split(',') | |
if'saml-provider' in chunks[0]: | |
newawsrole = chunks[1] + ',' + chunks[0] | |
index = awsroles.index(awsrole) | |
awsroles.insert(index, newawsrole) | |
awsroles.remove(awsrole) | |
# If I have more than one role, ask the user which one they want, | |
# otherwise just proceed | |
print "" | |
if len(awsroles) > 1: | |
i = 0 | |
print "Please choose the role you would like to assume:" | |
for awsrole in awsroles: | |
print '[', i, ']: ', awsrole.split(',')[0] | |
i += 1 | |
print "Selection: ", | |
selectedroleindex = raw_input() | |
# Basic sanity check of input | |
if int(selectedroleindex) > (len(awsroles) - 1): | |
print 'You selected an invalid role index, please try again' | |
sys.exit(0) | |
role_arn = awsroles[int(selectedroleindex)].split(',')[0] | |
principal_arn = awsroles[int(selectedroleindex)].split(',')[1] | |
else: | |
role_arn = awsroles[0].split(',')[0] | |
principal_arn = awsroles[0].split(',')[1] | |
# Use the assertion to get an AWS STS token using Assume Role with SAML | |
#print boto.sts.regions() | |
conn = boto.sts.connect_to_region(region) | |
#print conn.__dict__ | |
token = conn.assume_role_with_saml(role_arn, principal_arn, assertion,None, int(session_timeout)) | |
# Write the AWS STS token into the AWS credential file | |
home = expanduser("~") | |
filename = home + awsconfigfile | |
# Read in the existing config file | |
config = ConfigParser.RawConfigParser() | |
config.read(filename) | |
# Put the credentials into a saml specific section instead of clobbering | |
# the default credentials | |
if not config.has_section(profilename): | |
config.add_section(profilename) | |
config.set(profilename,'output', outputformat) | |
config.set(profilename,'region', region) | |
config.set(profilename,'aws_access_key_id', token.credentials.access_key) | |
config.set(profilename,'aws_secret_access_key', token.credentials.secret_key) | |
config.set(profilename,'aws_session_token', token.credentials.session_token) | |
# Write the updated config file | |
with open(filename, 'w+') as configfile: | |
config.write(configfile) | |
# Give the user some basic info as to what has just happened | |
print '\n\n----------------------------------------------------------------' | |
print 'Your new access key pair has been stored in the AWS configuration file {0} under the saml profile.'.format(filename) | |
print 'Note that it will expire at {0}.'.format(token.credentials.expiration, '%Y-%M-%s') | |
print 'After this time, you may safely rerun this script to refresh your access key pair.' | |
print 'To use this credential, call the AWS CLI with the --profile option (e.g. aws --profile saml ec2 describe-instances).' | |
print '----------------------------------------------------------------\n\n' |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment