openstack
/
deb-python-pysaml2
Code Issues Proposed changes

Error reporting clarified: CRITICAL level for unexpected conditions in saml2test; ERROR and lower for conditions in the test target; various documentation improvements

This commit is contained in:
rhoerbe 2014-08-11 14:33:40 +02:00
parent 781fbe4393
commit a2a4698936
8 changed files with 248 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
doc/sp_test/internal.rst
View File

@ -24,7 +24,7 @@ These files should be stored outside the saml2test package to have a clean separ
Client (sp_test/__init__.py)
.........................
Its life cycle is responsible for following activites:
- read config files and command line argumants
- read config files and command line argumants (the test driver's config is "json_config")
- initialize the test driver IDP
- initialize a Conversation
- start the Conversion with .do_sequence_and_tests()
@ -33,35 +33,37 @@ Its life cycle is responsible for following activites:
Conversation (sp_test/base.py)
..............................
operation (oper)
Operation (oper)
................
- comprises an id, name, sequence and tests
- names oper in
- Comprises an id, name, sequence and tests
- Example: 'sp-00': {"name": 'Basic Login test', "sequence": [(Login, AuthnRequest, AuthnResponse, None)], "tests": {"pre": [], "post": []}
- Names a use case in STHREP
OPERATIONS
..........
- set of operations provided in sp_test
- can be listed with the -l command line option
sequence
Sequence
........
- A list of flows
- Example: see "sequence" item in operation dictionary
- Example: see "sequence" item in operation dict
test (in the context of an operation)
Test (in the context of an operation)
....
- class to be executed as part of an operation, either before ("pre") or after ("post") the sequence
- class to be executed as part of an operation, either before ("pre") or after ("post") the sequence or inbetween a SAML request and response ("mid").
There are standard tests with the Request class (VerifyAuthnRequest) and operation-specific tests.
- Example for an operation-specific "mid" test: VerifyIfRequestIsSigned
- A test may be specified together with an argument as a tupel
flow
Flow
....
- A tupel of classes that together implement an SAML request-response pair between IDP and SP (and possible a discovery service). A class can be derived from Request, Response or (other), Check or Operation.
- A flow for a solicited authentication consists of 4 classes:
flow[0]: Operation (Handling a login flow such as discovery or WAYF - not implemented yet)
flow[1]: Request (process the authentication request)
flow[2]: Response (send the authentication response)
flow[3]: Check (optional - can be None. E.g. check the response if a correct error status was raised when sending a broken response)
* A tupel of classes that together implement an SAML request-response pair between IDP and SP (and possible other actors, such as a discovery service or IDP-proxy). A class can be derived from Request, Response (or other), Check or Operation.
* A flow for a solicited authentication consists of 4 classes:
* flow[0]: Operation (Handling a login flow such as discovery or WAYF - not implemented yet)
* flow[1]: Request (process the authentication request)
* flow[2]: Response (send the authentication response)
* flow[3]: Check (optional - can be None. E.g. check the response if a correct error status was raised when sending a broken response)
Check (and subclasses)
.....
@ -69,18 +71,17 @@ Check (and subclasses)
- writes a structured test report to conv.test_output
- It can check for expected errors, which do not cause an exception but in contrary are reported as success
interaction
Interaction
...........
- An interaction automates a human interaction. It searches a response from a test target for some constants, and if
there is a match, it will create a response suitable response.
(2) Simplefied Flow
(2) Simplyfied Flow
:::::::::::::::::::
The following pseudocdoe is an extract showing an overview of what is executed
for test sp-00:
The following pseudocode is an extract showing an overview of what is executed
for test sp-00::
::
do_sequence_and_test(self, oper, test):
self.test_sequence(tests["pre"]) # currently no tests defined for sp_test
for flow in oper:
@ -97,8 +98,8 @@ for test sp-00:
else:
self.handle_result()
send_idp_response(req, resp):
self.test_sequence(req.tests["post"]) # execute "post"-tests (request has "VerifyContent"-test built in; others from config)
send_idp_response(req_flow, resp_flow):
self.test_sequence(req_flow.tests["mid"]) # execute "mid"-tests (request has "VerifyContent"-test built in; others from config)
# this line stands for a part that is a bit more involved .. see source
args.update(resp._response_args) # set userid, identity
@ -129,3 +130,20 @@ for test sp-00:
_txt = self.last_response.content
if self.last_response.status_code >= 400:
raise FatalError("Did not expected error")
(3) Status Reporting
::::::::::::::::::::
The proper reporting of results is at the core of saml2test. Several conditions
must be considered:
#. An operation that was successful because the test target reports OK (e.g. HTTP 200)
#. An operation that was successful because the test target reports NOK as expected, e.g. because of an invalid signature - HTTP 500 could be the correct response
#. An errror in SAML2Test
#. An errror in configuration of SAML2Test
Status values are defined in saml2test.check like this:
INFORMATION = 0, OK = 1, WARNING = 2, ERROR = 3, CRITICAL = 4, INTERACTION = 5
There are 2 targets to write output to:
* Test_ouput is written to conv.test_ouput during the execution of the flows.

2
setup.py
View File

@ -26,7 +26,7 @@ setup(
author = "Roland Hedberg",
author_email = "roland.hedberg@adm.umu.se",
license="Apache 2.0",
packages=["idp_test", "idp_test/package", "saml2test", "sp_test"],
packages=["idp_test", "idp_test/package", "saml2test", "sp_test", "utility"],
package_dir = {"": "src"},
classifiers = [
"Development Status :: 4 - Beta",

22
src/saml2test/check.py
View File

@ -9,8 +9,8 @@ import sys
INFORMATION = 0
OK = 1
WARNING = 2
ERROR = 3
CRITICAL = 4
ERROR = 3 # an error condition in the test target
CRITICAL = 4 # an error condition in the test driver
INTERACTION = 5
STATUSCODE = ["INFORMATION", "OK", "WARNING", "ERROR", "CRITICAL",
@ -124,8 +124,8 @@ class CheckErrorResponse(ExpectedError):
class VerifyBadRequestResponse(ExpectedError):
"""
Verifies that the OP returned a 400 Bad Request response containing a
Error message.
Verifies that the test target returned a 400 Bad Request response
containing a an error message.
"""
cid = "verify-bad-request-response"
msg = "OP error"
@ -138,7 +138,7 @@ class VerifyBadRequestResponse(ExpectedError):
pass
else:
self._message = "Expected a 400 error message"
self._status = CRITICAL
self._status = ERROR
return res
@ -191,22 +191,22 @@ class Other(CriticalError):
msg = "Other error"
class CheckHTTPResponse(CriticalError):
class CheckSpHttpResponseOK(Error):
"""
Checks that the HTTP response status is within the 200 or 300 range
Checks that the SP's HTTP response status is within the 200 or 300 range
"""
cid = "check-http-response"
msg = "OP error"
cid = "check-sp-http-response-ok"
msg = "SP error OK"
def _func(self, conv):
_response = conv.last_response
_content = conv.last_content
_content = conv.last_response.content
res = {}
if _response.status_code >= 400:
self._status = self.status
self._message = self.msg
res["content"] = _content
#res["content"] = _content #too big + charset converstion needed
res["url"] = conv.position
res["http_status"] = _response.status_code

4
src/sp_test/__init__.py
View File

@ -74,8 +74,8 @@ class Client(object):
help="Path to the configuration file for the IdP")
self._parser.add_argument("-t", dest="testpackage",
help="Module describing tests")
self._parser.add_argument('-v', dest='verbose', action='store_true',
help="Print runtime information")
#self._parser.add_argument('-v', dest='verbose', action='store_true',
# help="Print runtime information") # unsused
self._parser.add_argument("-Y", dest="pysamllog", action='store_true',
help="Print PySAML2 logs")
self._parser.add_argument("oper", nargs="?", help="Which test to run")

61
src/sp_test/base.py
View File

@ -37,13 +37,13 @@ camel2underscore = re.compile('((?<=[a-z0-9])[A-Z]|(?!^)[A-Z](?=[a-z]))')
class Conversation():
def __init__(self, instance, config, interaction, json_config,
check_factory, entity_id, msg_factory=None,
features=None, verbose=False, constraints=None,
expect_exception=None):
features=None, constraints=None, # verbose=False,
expect_exception=None, commandlineargs=None):
self.instance = instance
self._config = config
self.test_output = []
self.features = features
self.verbose = verbose
#self.verbose = verbose # removed (not used)
self.check_factory = check_factory
self.msg_factory = msg_factory
self.expect_exception = expect_exception
@ -70,7 +70,7 @@ class Conversation():
self.position = ""
self.response = None
self.oper = None
self.idp_constraints = constraints
self.msg_constraints = constraints
self.json_config = json_config
self.start_page = json_config["start_page"]
@ -122,7 +122,7 @@ class Conversation():
kwargs = {}
self.do_check(test, **kwargs)
if test == ExpectedError:
return False
return False # TODO: return value is unused
return True
def my_endpoints(self):
@ -167,11 +167,6 @@ class Conversation():
elif isinstance(response(), Check):
self.do_check(response)
else:
# rhoerbe: I guess that this branch is never used, therefore
# I am proposing this exception:
#raise FatalError("can't use " + response.__class__.__name__ +
# ", because it is not a subclass of 'Check'")
#
# A HTTP redirect or HTTP Post
if 300 < self.last_response.status_code <= 303:
self._redirect(self.last_response)
@ -241,11 +236,11 @@ class Conversation():
break
return url
def send_idp_response(self, req, resp):
def send_idp_response(self, req_flow, resp_flow):
"""
:param req: The expected request
:param resp: The response type to be used
:return: A response
:param req_flow: The flow to check the request
:param resp_flow: The flow to prepare the response
:return: The SP's HTTP response on receiving the SAML response
"""
# make sure I got the request I expected
assert isinstance(self.saml_request.message, req._class)
@ -257,8 +252,8 @@ class Conversation():
# Pick information from the request that should be in the response
args = self.instance.response_args(self.saml_request.message,
[resp._binding])
_mods = list(resp.__mro__[:])
[resp_flow._binding])
_mods = list(resp_flow.__mro__[:])
_mods.reverse()
for m in _mods:
try:
@ -266,28 +261,32 @@ class Conversation():
except KeyError:
pass
args.update(resp._response_args)
args.update(resp_flow._response_args)
for param in ["identity", "userid"]:
if param in self.json_config:
args[param] = self.json_config[param]
if resp == ErrorResponse:
if resp_flow == ErrorResponse:
func = getattr(self.instance, "create_error_response")
else:
_op = camel2underscore.sub(r'_\1', req._class.c_tag).lower()
_op = camel2underscore.sub(r'_\1', req_flow._class.c_tag).lower()
func = getattr(self.instance, "create_%s_response" % _op)
# get from config which parts shall be signed
sign = []
for styp in ["sign_assertion", "sign_response"]:
if styp in args:
if args[styp].lower() == "always":
sign.append(styp)
del args[styp]
try:
if args[styp].lower() == "always":
sign.append(styp)
del args[styp]
except (AttributeError, TypeError):
raise AssertionError('config parameters "sign_assertion", '
'"sign_response" must be of type string')
response = func(**args)
response = resp(self).pre_processing(response)
response = resp_flow(self).pre_processing(response)
# and now for signing
if sign:
to_sign = []
@ -315,21 +314,21 @@ class Conversation():
response = signed_instance_factory(response, self.instance.sec,
to_sign)
info = self.instance.apply_binding(resp._binding, response,
info = self.instance.apply_binding(resp_flow._binding, response,
args["destination"],
self.relay_state,
"SAMLResponse", resp._sign)
"SAMLResponse", resp_flow._sign)
if resp._binding == BINDING_HTTP_REDIRECT:
if resp_flow._binding == BINDING_HTTP_REDIRECT:
url = None
for param, value in info["headers"]:
if param == "Location":
url = value
break
self.last_response = self.instance.send(url)
elif resp._binding == BINDING_HTTP_POST:
resp = base64.b64encode("%s" % response)
info["data"] = urllib.urlencode({"SAMLResponse": resp,
elif resp_flow._binding == BINDING_HTTP_POST:
resp_flow = base64.b64encode("%s" % response)
info["data"] = urllib.urlencode({"SAMLResponse": resp_flow,
"RelayState": self.relay_state})
info["method"] = "POST"
info["headers"] = {
@ -343,7 +342,7 @@ class Conversation():
Solicited or 'un-solicited' flows.
Solicited always starts with the Web client accessing a page.
Un-solicited starts with the IDP sending something.
Un-solicited starts with the IDP sending a SAMl Response.
"""
if len(flow) >= 3:
self.wb_send_GET_startpage()
@ -373,7 +372,7 @@ class Conversation():
break
except FatalError:
raise
except Exception:
except Exception as err:
#self.err_check("exception", err)
raise

145
src/sp_test/check.py
View File

@ -1,7 +1,9 @@
import inspect
import logging
import re
import sys
from saml2 import BINDING_HTTP_REDIRECT
from saml2test.check import Check
from saml2test.check import CRITICAL
from saml2test import check
@ -9,27 +11,23 @@ from saml2test.interaction import Interaction
__author__ = 'rolandh'
logger = logging.getLogger(__name__)
class VerifyContent(Check):
""" Basic content verification class, does required and max/min checks
class VerifyAuthnRequest(Check):
""" Basic AuthnRequest verification as provided by pysaml2
"""
cid = "verify-content"
cid = "verify-authnrequest"
def _func(self, conv):
try:
conv.saml_request.message.verify()
except ValueError:
self._status = CRITICAL
self._status = ERROR
return {}
class VerifyAuthnRequest(VerifyContent):
""" Basic AuthnRequest verification as provided by pysaml2
"""
cid = "verify-authnrequest"
class MatchResult(Check):
cid = "match-result"
@ -42,14 +40,131 @@ class MatchResult(Check):
class ErrorResponse(Check):
cid = "saml-error"
msg = "Expected error message"
msg = "Expected error message, but test target returned OK"
def _func(self, conv):
try:
assert conv.last_response.status_code >= 400
except AssertionError:
self._message = self.msg
self._status = CRITICAL
self._status = ERROR
return {}
class VerifyDigestAlgorithm(Check):
"""
verify that the used digest algorithm was one from the approved set.
"""
cid = "verify-digest-algorithm"
def _digest_algo(self, signature, allowed):
_alg = signature.signed_info.reference[0].digest_method.algorithm
try:
assert alg in allowed
except AssertionError:
self._message = "signature digest algorithm not allowed: " + alg
self._status = ERROR
return False
return True
def _func(self, conv):
if "digest_algorithm" not in conv.msg_constraints:
logger.info("Not verifying digest_algorithm (not configured)")
return {}
else:
try:
assert len(conv.msg_constraints["digest_algorithm"]) > 0
except AssertionError:
self._message = "List of allowed digest algorithm must not be empty"
self._status = ERROR
return {}
_algs = conv.msg_constraints["digest_algorithm"]
request = conv.saml_request.message
if request.signature:
if not self._digest_algo(request.signature, _algs):
return {}
return {}
class VerifyIfRequestIsSigned(Check):
"""
verify that the request has been signed
"""
cid = "verify-if-request-is-signed"
def _func(self, conv):
try:
check_sig = conv.msg_constraints["authnRequest_signature_required"]
except KeyError:
check_sig = False
if check_sig:
if conv._binding == BINDING_HTTP_REDIRECT:
try:
assert conv.http_parameters.signature is not None
except AssertionError:
self._message = "No AuthnRequest simple signature found"
self._status = ERROR
return {}
else:
try:
assert conv.saml_request.message.signature is not None
except AssertionError:
self._message = "No AuthnRequest XML signature found"
self._status = ERROR
return {}
else:
logger.debug("AuthnRequest signature is optional")
return {}
return {}
class VerifySignatureAlgorithm(Check):
"""
verify that the used signature algorithm was one from an approved set.
"""
cid = "verify-signature-algorithm"
def _func(self, conv):
if "signature_algorithm" not in conv.msg_constraints:
logger.info("Not verifying signature_algorithm (not configured)")
return {}
else:
try:
assert len(conv.msg_constraints["signature_algorithm"]) > 0
except AssertionError:
self._message = "List of allowed signature algorithm must " \
"not be empty"
self._status = ERROR
return {}
allowed_algs = [a[1] for a in conv.msg_constraints["signature_algorithm"]]
if conv._binding == BINDING_HTTP_REDIRECT:
if getattr(conv.http_parameters, "signature", None):
_alg = conv.http_parameters.sigalg
try:
assert _alg in allowed_algs
except AssertionError:
self._message = "Algorithm not in white list for " \
"redirect signing: " + _alg
self._status = ERROR
else:
signature = getattr(conv.saml_request.message, "signature", None)
if signature:
try:
assert signature.signed_info.signature_method.algorithm in \
allowed_algs
except AssertionError:
self._message = "Wrong algorithm used for signing: '%s'" % \
signature.signed_info.signature_method.algorithm
self._status = ERROR
else:
self._message = "cannot verify signature algorithm: request not signed"
self._status = WARNING
return {}
return {}
@ -70,7 +185,7 @@ class VerifyEchopageContents(Check):
except AssertionError:
self._message = "Cannot match expected static contents " \
"in SP echo page"
self._status = CRITICAL
self._status = ERROR
for pattern in conv.json_config["echopageContentPattern"]:
m = re.search(pattern, conv.last_response.content)
try:
@ -78,11 +193,11 @@ class VerifyEchopageContents(Check):
except AssertionError:
self._message = 'Cannot match expected response value' \
', pattern="' + pattern + '"'
self._status = CRITICAL
self._status = ERROR
except KeyError:
self._message = 'Configuration error: missing key ' \
'"echopageIdString" in test target config'
self._status = CRITICAL
self._status = ERROR
return {}
def call_on_redirect(self):

16
tests/idp_test/target_idp.py
View File

@ -4,6 +4,8 @@ from saml2.saml import NAME_FORMAT_URI
__author__ = 'rolandh'
import json
import xmldsig as ds
from saml2.saml import NAME_FORMAT_UNSPECIFIED, NAME_FORMAT_URI, NAME_FORMAT_BASIC
BASE = "http://localhost:8088"
@ -70,18 +72,24 @@ info = {
}
}
],
# metadata source for the test target's EntityDescriptor:
"metadata": metadata,
"name_format": NAME_FORMAT_URI
"constraints": {
"signature_algorithm": [ # allowed for assertion & response signature
ds.SIG_RSA_SHA1,
# test if attribute name format matches the given value. Absence of this
# option or the value NAME_FORMAT_UNSPECIFIED will match any format
#"name_format": NAME_FORMAT_BASIC,
#"name_format": NAME_FORMAT_UNSPECIFIED,
"name_format": NAME_FORMAT_URI,
# allowed for assertion & response:
"signature_algorithm": [
#ds.SIG_RSA_SHA1, # you may need this for legacy deployments
ds.SIG_RSA_SHA224,
ds.SIG_RSA_SHA256,
ds.SIG_RSA_SHA384,
ds.SIG_RSA_SHA512,
],
"digest_algorithm": [
ds.DIGEST_SHA1,
#ds.DIGEST_SHA1, # you may need this for legacy deployments
ds.DIGEST_SHA224,
ds.DIGEST_SHA256,
ds.DIGEST_SHA384,

22
tests/sp_test/targetsp.py
View File

@ -49,7 +49,27 @@ info = {
"echopageContentPattern": [r"Given Name\s*</td>\s*<td>Roland</td>",
r"Userid\s*</td>\s*<td>roalnd</td>",
r"Surname\s*</td>\s*<td>Hedberg</td>",
]
],
"constraints": {
"authnRequest_signature_required": True,
# allowed for assertion & response signature:
"signature_algorithm": [
#ds.SIG_RSA_SHA1, # you may need this for legacy deployments
ds.SIG_RSA_SHA224,
ds.SIG_RSA_SHA256,
ds.SIG_RSA_SHA384,
ds.SIG_RSA_SHA512,
],
"digest_algorithm": [
#ds.DIGEST_SHA1, # you may need this for legacy deployments
ds.DIGEST_SHA1,
ds.DIGEST_SHA224,
ds.DIGEST_SHA256,
ds.DIGEST_SHA384,
ds.DIGEST_SHA512,
ds.DIGEST_RIPEMD160,
],
},
}
print json.dumps(info)