| |
@@ -1,14 +1,11 @@
|
| |
# -*- coding: utf-8 -*-
|
| |
# SPDX-License-Identifier: MIT
|
| |
- from os import path, environ
|
| |
+ from os import path
|
| |
|
| |
import pytest
|
| |
import requests
|
| |
import mock
|
| |
from mock import patch, PropertyMock, Mock
|
| |
- import kerberos
|
| |
- import ldap3
|
| |
- from werkzeug.exceptions import Unauthorized as FlaskUnauthorized
|
| |
|
| |
import module_build_service.auth
|
| |
import module_build_service.errors
|
| |
@@ -220,209 +217,29 @@
|
| |
module_build_service.auth.get_user(request)
|
| |
assert str(cm.value) == "OIDC_REQUIRED_SCOPE must be set in server config."
|
| |
|
| |
+ @pytest.mark.parametrize("remote_username", ["", None])
|
| |
+ def test_get_user_kerberos_unauthorized(self, remote_username):
|
| |
+ request = Mock()
|
| |
+ request.environ.get.return_value = remote_username
|
| |
|
| |
- class KerberosMockConfig(object):
|
| |
- def __init__(
|
| |
- self,
|
| |
- uri="ldaps://test.example.local:636",
|
| |
- dn="ou=groups,dc=domain,dc=local",
|
| |
- kt="/path/to/keytab",
|
| |
- host="mbs.domain.local",
|
| |
- ):
|
| |
- """
|
| |
- :param uri: a string overriding config.ldap_uri
|
| |
- :param dn: a string overriding config.ldap_groups_dn
|
| |
- :param kt: a string overriding config.kerberos_keytab
|
| |
- :param host: a string overriding config.kerberos_http_host
|
| |
- """
|
| |
- self.uri = uri
|
| |
- self.dn = dn
|
| |
- self.kt = kt
|
| |
- self.host = host
|
| |
-
|
| |
- def __enter__(self):
|
| |
- self.auth_method_p = patch.object(
|
| |
- mbs_config.Config, "auth_method", new_callable=PropertyMock)
|
| |
- mocked_auth_method = self.auth_method_p.start()
|
| |
- mocked_auth_method.return_value = "kerberos"
|
| |
-
|
| |
- self.ldap_uri_p = patch.object(mbs_config.Config, "ldap_uri", new_callable=PropertyMock)
|
| |
- mocked_ldap_uri = self.ldap_uri_p.start()
|
| |
- mocked_ldap_uri.return_value = self.uri
|
| |
-
|
| |
- self.ldap_dn_p = patch.object(
|
| |
- mbs_config.Config, "ldap_groups_dn", new_callable=PropertyMock)
|
| |
- mocked_ldap_dn = self.ldap_dn_p.start()
|
| |
- mocked_ldap_dn.return_value = self.dn
|
| |
-
|
| |
- self.kerberos_keytab_p = patch.object(
|
| |
- mbs_config.Config, "kerberos_keytab", new_callable=PropertyMock)
|
| |
- mocked_kerberos_keytab = self.kerberos_keytab_p.start()
|
| |
- mocked_kerberos_keytab.return_value = self.kt
|
| |
-
|
| |
- self.kerberos_http_host_p = patch.object(
|
| |
- mbs_config.Config, "kerberos_http_host", new_callable=PropertyMock)
|
| |
- mocked_kerberos_http_host = self.kerberos_http_host_p.start()
|
| |
- mocked_kerberos_http_host.return_value = self.host
|
| |
-
|
| |
- def __exit__(self, *args):
|
| |
- self.auth_method_p.stop()
|
| |
- self.ldap_uri_p.stop()
|
| |
- self.ldap_dn_p.stop()
|
| |
- self.kerberos_keytab_p.stop()
|
| |
- self.kerberos_http_host_p.stop()
|
| |
-
|
| |
-
|
| |
- class TestAuthModuleKerberos:
|
| |
- @pytest.mark.parametrize("allowed_users", (set(), {"mprahl"}))
|
| |
- @patch("kerberos.authGSSServerInit", return_value=(kerberos.AUTH_GSS_COMPLETE, object()))
|
| |
- @patch("kerberos.authGSSServerStep", return_value=kerberos.AUTH_GSS_COMPLETE)
|
| |
- @patch("kerberos.authGSSServerResponse", return_value="STOKEN")
|
| |
- @patch("kerberos.authGSSServerUserName", return_value="mprahl@EXAMPLE.ORG")
|
| |
- @patch("kerberos.authGSSServerClean")
|
| |
- @patch("kerberos.getServerPrincipalDetails")
|
| |
- @patch.dict("os.environ")
|
| |
- @patch("module_build_service.auth.stack")
|
| |
- @patch.object(mbs_config.Config, "allowed_users", new_callable=PropertyMock)
|
| |
- def test_get_user_kerberos(
|
| |
- self, m_allowed_users, stack, principal, clean, name, response, step, init, allowed_users
|
| |
- ):
|
| |
- """
|
| |
- Test that authentication works with Kerberos and LDAP
|
| |
- """
|
| |
- m_allowed_users.return_value = allowed_users
|
| |
- mock_top = Mock()
|
| |
- stack.return_value = mock_top
|
| |
+ with pytest.raises(module_build_service.errors.Unauthorized):
|
| |
+ module_build_service.auth.get_user_kerberos(request)
|
| |
|
| |
- headers = {"Authorization": "foobar"}
|
| |
- request = mock.MagicMock()
|
| |
- request.headers.return_value = mock.MagicMock(spec_set=dict)
|
| |
- request.headers.__getitem__.side_effect = headers.__getitem__
|
| |
- request.headers.__setitem__.side_effect = headers.__setitem__
|
| |
- request.headers.__contains__.side_effect = headers.__contains__
|
| |
-
|
| |
- # Create the mock LDAP instance
|
| |
- server = ldap3.Server("ldaps://test.domain.local")
|
| |
- connection = ldap3.Connection(server, client_strategy=ldap3.MOCK_SYNC)
|
| |
- base_dn = "dc=domain,dc=local"
|
| |
- factory_group_attrs = {
|
| |
- "objectClass": ["top", "posixGroup"],
|
| |
- "memberUid": ["mprahl", "tbrady"],
|
| |
- "gidNumber": 1234,
|
| |
- "cn": ["factory2-devs"],
|
| |
- }
|
| |
- devs_group_attrs = {
|
| |
- "objectClass": ["top", "posixGroup"],
|
| |
- "memberUid": ["mprahl", "mikeb"],
|
| |
- "gidNumber": 1235,
|
| |
- "cn": ["devs"],
|
| |
- }
|
| |
- athletes_group_attrs = {
|
| |
- "objectClass": ["top", "posixGroup"],
|
| |
- "memberUid": ["tbrady", "rgronkowski"],
|
| |
- "gidNumber": 1236,
|
| |
- "cn": ["athletes"],
|
| |
- }
|
| |
- mprahl_attrs = {
|
| |
- "memberOf": ["cn=Employee,ou=groups,{0}".format(base_dn)],
|
| |
- "uid": ["mprahl"],
|
| |
- "cn": ["mprahl"],
|
| |
- "objectClass": ["top", "person"],
|
| |
- }
|
| |
- connection.strategy.add_entry(
|
| |
- "cn=factory2-devs,ou=groups,{0}".format(base_dn), factory_group_attrs
|
| |
- )
|
| |
- connection.strategy.add_entry(
|
| |
- "cn=athletes,ou=groups,{0}".format(base_dn), athletes_group_attrs
|
| |
- )
|
| |
- connection.strategy.add_entry("cn=devs,ou=groups,{0}".format(base_dn), devs_group_attrs)
|
| |
- connection.strategy.add_entry("cn=mprahl,ou=users,{0}".format(base_dn), mprahl_attrs)
|
| |
-
|
| |
- # If the user is in allowed_users, then group membership is not checked, and an empty set
|
| |
- # is just returned for the groups
|
| |
- if allowed_users:
|
| |
- expected_groups = set()
|
| |
- else:
|
| |
- expected_groups = {"devs", "factory2-devs"}
|
| |
-
|
| |
- with patch("ldap3.Connection") as mock_ldap_con, KerberosMockConfig():
|
| |
- mock_ldap_con.return_value = connection
|
| |
- assert module_build_service.auth.get_user_kerberos(request) == (
|
| |
- "mprahl", expected_groups)
|
| |
-
|
| |
- def test_auth_header_not_set(self):
|
| |
- """
|
| |
- Test that an Unauthorized exception is returned when there is no authorization header
|
| |
- set.
|
| |
- """
|
| |
- headers = {}
|
| |
- request = mock.MagicMock()
|
| |
- request.headers.return_value = mock.MagicMock(spec_set=dict)
|
| |
- request.headers.__getitem__.side_effect = headers.__getitem__
|
| |
- request.headers.__setitem__.side_effect = headers.__setitem__
|
| |
- request.headers.__contains__.side_effect = headers.__contains__
|
| |
-
|
| |
- with KerberosMockConfig():
|
| |
- try:
|
| |
- module_build_service.auth.get_user_kerberos(request)
|
| |
- assert False, "Unauthorized error not raised"
|
| |
- except FlaskUnauthorized as error:
|
| |
- assert error.response.www_authenticate.to_header().strip() == "Negotiate"
|
| |
- assert error.response.status == "401 UNAUTHORIZED"
|
| |
-
|
| |
- @patch.dict(environ)
|
| |
- def test_keytab_not_set(self):
|
| |
- """
|
| |
- Test that authentication fails when the keytab is not set
|
| |
- """
|
| |
- if "KRB5_KTNAME" in environ:
|
| |
- del environ["KRB5_KTNAME"]
|
| |
-
|
| |
- headers = {"Authorization": "foobar"}
|
| |
- request = mock.MagicMock()
|
| |
- request.headers.return_value = mock.MagicMock(spec_set=dict)
|
| |
- request.headers.__getitem__.side_effect = headers.__getitem__
|
| |
- request.headers.__setitem__.side_effect = headers.__setitem__
|
| |
- request.headers.__contains__.side_effect = headers.__contains__
|
| |
-
|
| |
- with KerberosMockConfig(kt=""):
|
| |
- try:
|
| |
- module_build_service.auth.get_user_kerberos(request)
|
| |
- assert False, "Unauthorized error not raised"
|
| |
- except module_build_service.errors.Unauthorized as error:
|
| |
- assert str(error) == (
|
| |
- 'Kerberos: set the config value of "KERBEROS_KEYTAB" '
|
| |
- 'or the environment variable "KRB5_KTNAME" to your keytab file'
|
| |
- )
|
| |
+ @patch.object(module_build_service.auth.conf, "allowed_users", new=["someone", "somebody"])
|
| |
+ def test_get_user_kerberos_user_is_in_allowed_users_group(self):
|
| |
+ request = Mock()
|
| |
+ request.environ.get.return_value = "someone@realm"
|
| |
|
| |
- # Set the return value to something not 0 (continue) or 1 (complete)
|
| |
- @patch("kerberos.authGSSServerInit", return_value=(100, object()))
|
| |
- @patch("kerberos.authGSSServerStep", return_value=kerberos.AUTH_GSS_COMPLETE)
|
| |
- @patch("kerberos.authGSSServerResponse", return_value="STOKEN")
|
| |
- @patch("kerberos.authGSSServerUserName", return_value="mprahl@EXAMPLE.ORG")
|
| |
- @patch("kerberos.authGSSServerClean")
|
| |
- @patch("kerberos.getServerPrincipalDetails")
|
| |
- @patch.dict("os.environ")
|
| |
- @patch("module_build_service.auth.stack")
|
| |
- def test_get_user_kerberos_invalid_ticket(
|
| |
- self, stack, principal, clean, name, response, step, init
|
| |
- ):
|
| |
- """
|
| |
- Test that authentication fails with an invalid Kerberos ticket
|
| |
- """
|
| |
- mock_top = Mock()
|
| |
- stack.return_value = mock_top
|
| |
-
|
| |
- headers = {"Authorization": "foobar"}
|
| |
- request = mock.MagicMock()
|
| |
- request.headers.return_value = mock.MagicMock(spec_set=dict)
|
| |
- request.headers.__getitem__.side_effect = headers.__getitem__
|
| |
- request.headers.__setitem__.side_effect = headers.__setitem__
|
| |
- request.headers.__contains__.side_effect = headers.__contains__
|
| |
-
|
| |
- with KerberosMockConfig():
|
| |
- try:
|
| |
- module_build_service.auth.get_user_kerberos(request)
|
| |
- assert False, "Forbidden error not raised"
|
| |
- except module_build_service.errors.Forbidden as error:
|
| |
- assert str(error) == ("Invalid Kerberos ticket")
|
| |
+ username, groups = module_build_service.auth.get_user_kerberos(request)
|
| |
+ assert "someone" == username
|
| |
+ assert set() == groups
|
| |
+
|
| |
+ @patch.object(module_build_service.auth.conf, "allowed_users", new=["someone", "somebody"])
|
| |
+ @patch("module_build_service.auth.get_ldap_group_membership", return_value=["group1", "group2"])
|
| |
+ def test_get_user_kerberos_user_is_not_in_allowed_users_group(self, get_ldap_group_membership):
|
| |
+ request = Mock()
|
| |
+ request.environ.get.return_value = "x-man@realm"
|
| |
+
|
| |
+ username, groups = module_build_service.auth.get_user_kerberos(request)
|
| |
+ assert "x-man" == username
|
| |
+ assert {"group1", "group2"} == groups
|
| |
cqi
commented 4 years ago
This change will be tested with changes to MBS ansible playbook.
Resolves: FACTORY-3894