mirror of
https://pagure.io/fm-orchestrator.git
synced 2026-04-24 10:43:16 +08:00
Remove KerberosAuthenticate and get kerberos username from REMOTE_USER
Signed-off-by: Chenxiong Qi <cqi@redhat.com>
This commit is contained in:
Chenxiong Qi
@@ -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 @@ class TestAuthModule:
|
||||
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
|
||||
with pytest.raises(module_build_service.errors.Unauthorized):
|
||||
module_build_service.auth.get_user_kerberos(request)
|
||||
|
||||
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"
|
||||
@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"
|
||||
|
||||
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
|
||||
username, groups = module_build_service.auth.get_user_kerberos(request)
|
||||
assert "someone" == username
|
||||
assert set() == groups
|
||||
|
||||
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
|
||||
@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"
|
||||
|
||||
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
|
||||
|
||||
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'
|
||||
)
|
||||
|
||||
# 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 "x-man" == username
|
||||
assert {"group1", "group2"} == groups
|
||||
|
||||
Reference in New Issue
Block a user