authentik.enterprise.providers.ws_federation.tests.test_sign_in

View Source

 1import xmlsec
 2from django.test import TestCase
 3from guardian.utils import get_anonymous_user
 4from lxml import etree  # nosec
 5
 6from authentik.core.models import Application
 7from authentik.core.tests.utils import RequestFactory, create_test_cert, create_test_flow
 8from authentik.enterprise.providers.ws_federation.models import WSFederationProvider
 9from authentik.enterprise.providers.ws_federation.processors.constants import (
10    NS_MAP,
11    WS_FED_ACTION_SIGN_IN,
12    WS_FED_POST_KEY_RESULT,
13)
14from authentik.enterprise.providers.ws_federation.processors.sign_in import (
15    SignInProcessor,
16    SignInRequest,
17)
18from authentik.lib.generators import generate_id
19from authentik.lib.xml import lxml_from_string
20
21
22class TestWSFedSignIn(TestCase):
23    def setUp(self):
24        self.flow = create_test_flow()
25        self.cert = create_test_cert()
26        self.provider = WSFederationProvider.objects.create(
27            name=generate_id(),
28            authorization_flow=self.flow,
29            signing_kp=self.cert,
30            acs_url="https://t.goauthentik.io",
31            audience="foo",
32        )
33        self.app = Application.objects.create(
34            name=generate_id(), slug=generate_id(), provider=self.provider
35        )
36        self.factory = RequestFactory()
37
38    def test_wreply(self):
39        request = self.factory.get(
40            "/?wreply=https://t.goauthentik.io/foo&wa=wsignin1.0&wtrealm=foo",
41            user=get_anonymous_user(),
42        )
43        SignInRequest.parse(request)
44        with self.assertRaises(ValueError):
45            request = self.factory.get(
46                "/?wreply=https://t.goauthentik.io.invalid.com&wa=wsignin1.0&wtrealm=foo",
47                user=get_anonymous_user(),
48            )
49            SignInRequest.parse(request)
50
51    def test_token_gen(self):
52        request = self.factory.get("/", user=get_anonymous_user())
53        proc = SignInProcessor(
54            self.provider,
55            request,
56            SignInRequest(
57                wa=WS_FED_ACTION_SIGN_IN,
58                wtrealm="",
59                wreply="",
60                wctx=None,
61            ),
62        )
63        token = proc.response()[WS_FED_POST_KEY_RESULT]
64
65        root = lxml_from_string(token)
66
67        schema = etree.XMLSchema(
68            etree.parse(source="schemas/ws-trust.xsd", parser=etree.XMLParser())  # nosec
69        )
70        self.assertTrue(schema.validate(etree=root), schema.error_log)
71
72    def test_signature(self):
73        request = self.factory.get("/", user=get_anonymous_user())
74        proc = SignInProcessor(
75            self.provider,
76            request,
77            SignInRequest(
78                wa=WS_FED_ACTION_SIGN_IN,
79                wtrealm="",
80                wreply="",
81                wctx=None,
82            ),
83        )
84        token = proc.response()[WS_FED_POST_KEY_RESULT]
85
86        root = lxml_from_string(token)
87        xmlsec.tree.add_ids(root, ["ID"])
88        signature_nodes = root.xpath("//saml:Assertion/ds:Signature", namespaces=NS_MAP)
89        self.assertEqual(len(signature_nodes), 1)
90
91        signature_node = signature_nodes[0]
92        ctx = xmlsec.SignatureContext()
93        ctx.key = xmlsec.Key.from_memory(
94            self.cert.certificate_data,
95            xmlsec.constants.KeyDataFormatCertPem,
96            None,
97        )
98        ctx.verify(signature_node)

class TestWSFedSignIn(django.test.testcases.TestCase): View Source

23class TestWSFedSignIn(TestCase):
24    def setUp(self):
25        self.flow = create_test_flow()
26        self.cert = create_test_cert()
27        self.provider = WSFederationProvider.objects.create(
28            name=generate_id(),
29            authorization_flow=self.flow,
30            signing_kp=self.cert,
31            acs_url="https://t.goauthentik.io",
32            audience="foo",
33        )
34        self.app = Application.objects.create(
35            name=generate_id(), slug=generate_id(), provider=self.provider
36        )
37        self.factory = RequestFactory()
38
39    def test_wreply(self):
40        request = self.factory.get(
41            "/?wreply=https://t.goauthentik.io/foo&wa=wsignin1.0&wtrealm=foo",
42            user=get_anonymous_user(),
43        )
44        SignInRequest.parse(request)
45        with self.assertRaises(ValueError):
46            request = self.factory.get(
47                "/?wreply=https://t.goauthentik.io.invalid.com&wa=wsignin1.0&wtrealm=foo",
48                user=get_anonymous_user(),
49            )
50            SignInRequest.parse(request)
51
52    def test_token_gen(self):
53        request = self.factory.get("/", user=get_anonymous_user())
54        proc = SignInProcessor(
55            self.provider,
56            request,
57            SignInRequest(
58                wa=WS_FED_ACTION_SIGN_IN,
59                wtrealm="",
60                wreply="",
61                wctx=None,
62            ),
63        )
64        token = proc.response()[WS_FED_POST_KEY_RESULT]
65
66        root = lxml_from_string(token)
67
68        schema = etree.XMLSchema(
69            etree.parse(source="schemas/ws-trust.xsd", parser=etree.XMLParser())  # nosec
70        )
71        self.assertTrue(schema.validate(etree=root), schema.error_log)
72
73    def test_signature(self):
74        request = self.factory.get("/", user=get_anonymous_user())
75        proc = SignInProcessor(
76            self.provider,
77            request,
78            SignInRequest(
79                wa=WS_FED_ACTION_SIGN_IN,
80                wtrealm="",
81                wreply="",
82                wctx=None,
83            ),
84        )
85        token = proc.response()[WS_FED_POST_KEY_RESULT]
86
87        root = lxml_from_string(token)
88        xmlsec.tree.add_ids(root, ["ID"])
89        signature_nodes = root.xpath("//saml:Assertion/ds:Signature", namespaces=NS_MAP)
90        self.assertEqual(len(signature_nodes), 1)
91
92        signature_node = signature_nodes[0]
93        ctx = xmlsec.SignatureContext()
94        ctx.key = xmlsec.Key.from_memory(
95            self.cert.certificate_data,
96            xmlsec.constants.KeyDataFormatCertPem,
97            None,
98        )
99        ctx.verify(signature_node)

Similar to TransactionTestCase, but use transaction.atomic() to achieve test isolation.

In most situations, TestCase should be preferred to TransactionTestCase as it allows faster execution. However, there are some situations where using TransactionTestCase might be necessary (e.g. testing some transactional behavior).

On database backends with no transaction support, TestCase behaves as TransactionTestCase.

def setUp(self): View Source

24    def setUp(self):
25        self.flow = create_test_flow()
26        self.cert = create_test_cert()
27        self.provider = WSFederationProvider.objects.create(
28            name=generate_id(),
29            authorization_flow=self.flow,
30            signing_kp=self.cert,
31            acs_url="https://t.goauthentik.io",
32            audience="foo",
33        )
34        self.app = Application.objects.create(
35            name=generate_id(), slug=generate_id(), provider=self.provider
36        )
37        self.factory = RequestFactory()

Hook method for setting up the test fixture before exercising it.

def test_wreply(self): View Source

39    def test_wreply(self):
40        request = self.factory.get(
41            "/?wreply=https://t.goauthentik.io/foo&wa=wsignin1.0&wtrealm=foo",
42            user=get_anonymous_user(),
43        )
44        SignInRequest.parse(request)
45        with self.assertRaises(ValueError):
46            request = self.factory.get(
47                "/?wreply=https://t.goauthentik.io.invalid.com&wa=wsignin1.0&wtrealm=foo",
48                user=get_anonymous_user(),
49            )
50            SignInRequest.parse(request)

def test_token_gen(self): View Source

52    def test_token_gen(self):
53        request = self.factory.get("/", user=get_anonymous_user())
54        proc = SignInProcessor(
55            self.provider,
56            request,
57            SignInRequest(
58                wa=WS_FED_ACTION_SIGN_IN,
59                wtrealm="",
60                wreply="",
61                wctx=None,
62            ),
63        )
64        token = proc.response()[WS_FED_POST_KEY_RESULT]
65
66        root = lxml_from_string(token)
67
68        schema = etree.XMLSchema(
69            etree.parse(source="schemas/ws-trust.xsd", parser=etree.XMLParser())  # nosec
70        )
71        self.assertTrue(schema.validate(etree=root), schema.error_log)

def test_signature(self): View Source

73    def test_signature(self):
74        request = self.factory.get("/", user=get_anonymous_user())
75        proc = SignInProcessor(
76            self.provider,
77            request,
78            SignInRequest(
79                wa=WS_FED_ACTION_SIGN_IN,
80                wtrealm="",
81                wreply="",
82                wctx=None,
83            ),
84        )
85        token = proc.response()[WS_FED_POST_KEY_RESULT]
86
87        root = lxml_from_string(token)
88        xmlsec.tree.add_ids(root, ["ID"])
89        signature_nodes = root.xpath("//saml:Assertion/ds:Signature", namespaces=NS_MAP)
90        self.assertEqual(len(signature_nodes), 1)
91
92        signature_node = signature_nodes[0]
93        ctx = xmlsec.SignatureContext()
94        ctx.key = xmlsec.Key.from_memory(
95            self.cert.certificate_data,
96            xmlsec.constants.KeyDataFormatCertPem,
97            None,
98        )
99        ctx.verify(signature_node)