mirror of
https://salsa.debian.org/freeipa-team/freeipa.git
synced 2026-10-25 22:50:12 -05:00
Florence Blanc-Renaud
b0e4cdbf9d
The test is performing ssh from the runner to the master but is using the external_hostname and randomly fails. Make sure to use the configured hostname instead. Fixes: https://pagure.io/freeipa/issue/9849 Signed-off-by: Florence Blanc-Renaud <flo@redhat.com> Reviewed-By: Rob Crittenden <rcritten@redhat.com>
799 lines
32 KiB
Python
799 lines
32 KiB
Python
#
|
|
# Copyright (C) 2019 FreeIPA Contributors see COPYING for license
|
|
#
|
|
"""OTP token tests
|
|
"""
|
|
import base64
|
|
import logging
|
|
import re
|
|
import tempfile
|
|
import textwrap
|
|
import time
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import pytest
|
|
from cryptography.hazmat.backends import default_backend
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.twofactor.hotp import HOTP
|
|
from cryptography.hazmat.primitives.twofactor.totp import TOTP
|
|
from ldap.controls.simple import BooleanControl
|
|
from paramiko import AuthenticationException
|
|
|
|
from ipalib import errors
|
|
from ipaplatform.osinfo import osinfo
|
|
from ipaplatform.paths import paths
|
|
from ipapython.dn import DN
|
|
from ipatests.pytest_ipa.integration import tasks
|
|
from ipatests.test_integration.base import IntegrationTest
|
|
from ipatests.util import xfail_context
|
|
|
|
PASSWORD = "DummyPassword123"
|
|
USER = "opttestuser"
|
|
ARMOR = "/tmp/armor"
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def add_otptoken(host, owner, *, otptype="hotp", digits=6, algo="sha1"):
|
|
args = [
|
|
"ipa",
|
|
"otptoken-add",
|
|
"--owner",
|
|
owner,
|
|
"--type",
|
|
otptype,
|
|
"--digits",
|
|
str(digits),
|
|
"--algo",
|
|
algo,
|
|
"--no-qrcode",
|
|
]
|
|
result = host.run_command(args)
|
|
otpuid = re.search(
|
|
r"Unique ID:\s*([a-z0-9-]*)\s+", result.stdout_text
|
|
).group(1)
|
|
otpuristr = re.search(r"URI:\s*(.*)\s+", result.stdout_text).group(1)
|
|
otpuri = urlparse(otpuristr)
|
|
assert otpuri.netloc == otptype
|
|
|
|
query = parse_qs(otpuri.query)
|
|
assert query["algorithm"][0] == algo.upper()
|
|
assert query["digits"][0] == str(digits)
|
|
key = base64.b32decode(query["secret"][0])
|
|
assert len(key) == 35
|
|
|
|
hashcls = getattr(hashes, algo.upper())
|
|
if otptype == "hotp":
|
|
return otpuid, HOTP(key, digits, hashcls(), default_backend())
|
|
else:
|
|
period = int(query["period"][0])
|
|
return otpuid, TOTP(key, digits, hashcls(), period, default_backend())
|
|
|
|
|
|
def del_otptoken(host, otpuid):
|
|
tasks.kinit_admin(host)
|
|
host.run_command(["ipa", "otptoken-del", otpuid])
|
|
|
|
|
|
def kinit_otp(host, user, *, password, otp, success=True):
|
|
tasks.kdestroy_all(host)
|
|
# create armor for FAST
|
|
host.run_command(["kinit", "-n", "-c", ARMOR])
|
|
host.run_command(
|
|
["kinit", "-T", ARMOR, user],
|
|
stdin_text=f"{password}{otp}\n",
|
|
ok_returncode=0 if success else 1,
|
|
)
|
|
|
|
|
|
def ssh_2fa_with_cmd(host, hostname, username, password, otpvalue,
|
|
command="exit 0"):
|
|
""" ssh to user and in same session pass the command to check tgt of user
|
|
:param host: host to ssh
|
|
:param hostname: hostname to ssh
|
|
:param str username: The name of user
|
|
:param str password: password, usually the first factor
|
|
:param str otpvalue: generated pin of user
|
|
:param str command: command to execute in same session,
|
|
by deafult set to "exit 0"
|
|
:return: object class of expect command run
|
|
"""
|
|
temp_conf = tempfile.NamedTemporaryFile(suffix='.exp', delete=False)
|
|
with open(temp_conf.name, 'w') as tfile:
|
|
tfile.write('proc exitmsg { msg code } {\n')
|
|
tfile.write('\t# Close spawned program, if we are in the prompt\n')
|
|
tfile.write('\tcatch close\n\n')
|
|
tfile.write('\t# Wait for the exit code\n')
|
|
tfile.write('\tlassign [wait] pid spawnid os_error_flag rc\n\n')
|
|
tfile.write('\tputs ""\n')
|
|
tfile.write('\tputs "expect result: $msg"\n')
|
|
tfile.write('\tputs "expect exit code: $code"\n')
|
|
tfile.write('\tputs "expect spawn exit code: $rc"\n')
|
|
tfile.write('\texit $code\n')
|
|
tfile.write('}\n')
|
|
tfile.write('set timeout 60\n')
|
|
tfile.write('set prompt ".*\\[#\\$>\\] $"\n')
|
|
tfile.write(f'set password "{password}"\n')
|
|
tfile.write(f'set otpvalue "{otpvalue}"\n')
|
|
tfile.write(f'spawn ssh -o NumberOfPasswordPrompts=1 -o '
|
|
f'StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null'
|
|
f' -l {username} {hostname} {command}\n')
|
|
tfile.write('expect {\n')
|
|
tfile.write('"Enter first factor:*" {send -- "$password\r"}\n')
|
|
tfile.write('timeout {exitmsg "Unexpected output" 201}\n')
|
|
tfile.write('eof {exitmsg "Unexpected end of file" 202}\n')
|
|
tfile.write('}\n')
|
|
tfile.write('expect {\n')
|
|
tfile.write('"Enter second factor:*" {send -- "$otpvalue\r"}\n')
|
|
tfile.write('timeout {exitmsg "Unexpected output" 201}\n')
|
|
tfile.write('eof {exitmsg "Unexpected end of file" 202}\n')
|
|
tfile.write('}\n')
|
|
tfile.write('expect {\n')
|
|
tfile.write('"Authentication failure" '
|
|
'{exitmsg "Authentication failure" 1}\n')
|
|
tfile.write('eof {exitmsg "Authentication successful" 0}\n')
|
|
tfile.write('timeout {exitmsg "Unexpected output" 201}\n')
|
|
tfile.write('}\n')
|
|
tfile.write('expect {\n')
|
|
tfile.write('exitmsg "Unexpected code path" 203\n')
|
|
tfile.write('EOF\n')
|
|
tfile.write('}')
|
|
host.transport.put_file(temp_conf.name, '/tmp/ssh.exp')
|
|
tasks.clear_sssd_cache(host)
|
|
expect_cmd = 'expect -f /tmp/ssh.exp'
|
|
cmd = host.run_command(expect_cmd, raiseonerr=False)
|
|
return cmd
|
|
|
|
|
|
def ssh_2f(hostname, username, answers_dict, port=22, unwanted_prompt=""):
|
|
"""
|
|
:param hostname: hostname
|
|
:param username: username
|
|
:param answers_dict: dictionary of options with prompt_message and value.
|
|
:param port: port for ssh
|
|
"""
|
|
|
|
# Handler for server questions
|
|
def answer_handler(title, instructions, prompt_list):
|
|
resp = []
|
|
if title:
|
|
print(title.strip())
|
|
if instructions:
|
|
print(instructions.strip())
|
|
for prmpt in prompt_list:
|
|
prmpt_str = prmpt[0].strip()
|
|
resp.append(answers_dict[prmpt_str])
|
|
logger.info("Prompt is: '%s'", prmpt_str)
|
|
logger.info(
|
|
"Answer to ssh prompt is: '%s'", answers_dict[prmpt_str])
|
|
if unwanted_prompt and prmpt_str == unwanted_prompt:
|
|
# We should not see this prompt
|
|
raise ValueError("We got an unwanted prompt: "
|
|
+ answers_dict[prmpt_str])
|
|
return resp
|
|
|
|
import paramiko
|
|
trans = paramiko.Transport((hostname, port))
|
|
trans.connect()
|
|
trans.auth_interactive(username, answer_handler)
|
|
|
|
|
|
def set_sssd_conf(host, add_contents):
|
|
contents = host.get_file_contents(paths.SSSD_CONF, encoding="utf-8")
|
|
file_contents = contents + add_contents
|
|
host.put_file_contents(paths.SSSD_CONF, file_contents)
|
|
tasks.clear_sssd_cache(host)
|
|
|
|
|
|
class TestOTPToken(IntegrationTest):
|
|
"""Tests for member manager feature for groups and hostgroups
|
|
"""
|
|
|
|
topology = "line"
|
|
|
|
@classmethod
|
|
def install(cls, mh):
|
|
master = cls.master
|
|
tasks.install_packages(master, ['expect'])
|
|
super(TestOTPToken, cls).install(mh)
|
|
|
|
tasks.kinit_admin(master)
|
|
# create service with OTP auth indicator
|
|
cls.service_name = f"otponly/{master.hostname}"
|
|
master.run_command(
|
|
["ipa", "service-add", cls.service_name, "--auth-ind=otp"]
|
|
)
|
|
# service needs a keytab before user can acquire a ticket for it
|
|
keytab = "/tmp/otponly.keytab"
|
|
master.run_command(
|
|
["ipa-getkeytab", "-p", cls.service_name, "-k", keytab]
|
|
)
|
|
master.run_command(["rm", "-f", keytab])
|
|
|
|
tasks.create_active_user(master, USER, PASSWORD)
|
|
tasks.kinit_admin(master)
|
|
master.run_command(["ipa", "user-mod", USER, "--user-auth-type=otp"])
|
|
|
|
@classmethod
|
|
def uninstall(cls, mh):
|
|
cls.master.run_command(["rm", "-f", ARMOR])
|
|
super(TestOTPToken, cls).uninstall(mh)
|
|
|
|
def test_otp_auth_ind(self):
|
|
tasks.kinit_admin(self.master)
|
|
result = self.master.run_command(
|
|
["kvno", self.service_name], ok_returncode=1
|
|
)
|
|
assert "KDC policy rejects request" in result.stderr_text
|
|
|
|
def test_hopt(self):
|
|
master = self.master
|
|
|
|
tasks.kinit_admin(self.master)
|
|
otpuid, hotp = add_otptoken(master, USER, otptype="hotp")
|
|
master.run_command(["ipa", "otptoken-show", otpuid])
|
|
# normal password login fails
|
|
master.run_command(
|
|
["kinit", USER], stdin_text=f"{PASSWORD}\n", ok_returncode=1
|
|
)
|
|
# OTP login works
|
|
otpvalue = hotp.generate(0).decode("ascii")
|
|
kinit_otp(master, USER, password=PASSWORD, otp=otpvalue)
|
|
# repeating OTP fails
|
|
kinit_otp(
|
|
master, USER, password=PASSWORD, otp=otpvalue, success=False
|
|
)
|
|
# skipping an OTP is ok
|
|
otpvalue = hotp.generate(2).decode("ascii")
|
|
kinit_otp(master, USER, password=PASSWORD, otp=otpvalue)
|
|
# TGT with OTP auth indicator can get a ticket for OTP-only service
|
|
master.run_command(["kvno", self.service_name])
|
|
result = master.run_command(["klist"])
|
|
assert self.service_name in result.stdout_text
|
|
|
|
del_otptoken(master, otpuid)
|
|
|
|
@pytest.fixture
|
|
def desynchronized_hotp(self):
|
|
""" Create an hotp token for user """
|
|
tasks.kinit_admin(self.master)
|
|
otpuid, hotp = add_otptoken(self.master, USER, otptype="hotp")
|
|
|
|
# skipping too many OTP fails
|
|
otp1 = hotp.generate(10).decode("ascii")
|
|
kinit_otp(self.master, USER, password=PASSWORD, otp=otp1,
|
|
success=False)
|
|
# Now the token is desynchronized
|
|
yield (otpuid, hotp)
|
|
|
|
del_otptoken(self.master, otpuid)
|
|
|
|
def test_otptoken_sync_incorrect_password(self, desynchronized_hotp):
|
|
""" Test if sync fails when incorrect password is provided """
|
|
otpuid, hotp = desynchronized_hotp
|
|
|
|
otp2 = hotp.generate(20).decode("ascii")
|
|
otp3 = hotp.generate(21).decode("ascii")
|
|
|
|
# Try to sync with a wrong password
|
|
result = self.master.run_command(
|
|
["ipa", "otptoken-sync", "--user", USER, otpuid],
|
|
stdin_text=f"invalidpwd\n{otp2}\n{otp3}\n", raiseonerr=False
|
|
)
|
|
assert result.returncode == 1
|
|
assert "Invalid Credentials!" in result.stderr_text
|
|
|
|
# Now sync with the right values
|
|
self.master.run_command(
|
|
["ipa", "otptoken-sync", "--user", USER, otpuid],
|
|
stdin_text=f"{PASSWORD}\n{otp2}\n{otp3}\n"
|
|
)
|
|
|
|
def test_otptoken_sync_incorrect_first_value(self, desynchronized_hotp):
|
|
""" Test if sync fails when incorrect 1st token value is provided """
|
|
otpuid, hotp = desynchronized_hotp
|
|
|
|
otp2 = "12345a"
|
|
otp3 = hotp.generate(20).decode("ascii")
|
|
otp4 = hotp.generate(21).decode("ascii")
|
|
|
|
# Try to sync with a wrong first value (contains non-digit)
|
|
result = self.master.run_command(
|
|
["ipa", "otptoken-sync", "--user", USER, otpuid],
|
|
stdin_text=f"{PASSWORD}\n{otp2}\n{otp3}\n", raiseonerr=False
|
|
)
|
|
assert result.returncode == 1
|
|
assert "Invalid Credentials!" in result.stderr_text
|
|
|
|
# Now sync with the right values
|
|
self.master.run_command(
|
|
["ipa", "otptoken-sync", "--user", USER, otpuid],
|
|
stdin_text=f"{PASSWORD}\n{otp3}\n{otp4}\n"
|
|
)
|
|
|
|
def test_otptoken_sync_incorrect_second_value(self, desynchronized_hotp):
|
|
""" Test if sync fails when incorrect 2nd token value is provided """
|
|
otpuid, hotp = desynchronized_hotp
|
|
|
|
otp2 = hotp.generate(20).decode("ascii")
|
|
otp3 = hotp.generate(21).decode("ascii")
|
|
# Try to sync with wrong order
|
|
result = self.master.run_command(
|
|
["ipa", "otptoken-sync", "--user", USER, otpuid],
|
|
stdin_text=f"{PASSWORD}\n{otp3}\n{otp2}\n", raiseonerr=False
|
|
)
|
|
assert result.returncode == 1
|
|
assert "Invalid Credentials!" in result.stderr_text
|
|
|
|
# Now sync with the right order
|
|
self.master.run_command(
|
|
["ipa", "otptoken-sync", "--user", USER, otpuid],
|
|
stdin_text=f"{PASSWORD}\n{otp2}\n{otp3}\n"
|
|
)
|
|
|
|
def test_totp(self):
|
|
master = self.master
|
|
|
|
tasks.kinit_admin(self.master)
|
|
otpuid, totp = add_otptoken(master, USER, otptype="totp")
|
|
|
|
otpvalue = totp.generate(int(time.time())).decode("ascii")
|
|
kinit_otp(master, USER, password=PASSWORD, otp=otpvalue)
|
|
# TGT with OTP auth indicator can get a ticket for OTP-only service
|
|
master.run_command(["kvno", self.service_name])
|
|
result = master.run_command(["klist"])
|
|
assert self.service_name in result.stdout_text
|
|
|
|
del_otptoken(master, otpuid)
|
|
|
|
def test_otptoken_sync(self):
|
|
master = self.master
|
|
|
|
tasks.kinit_admin(self.master)
|
|
otpuid, hotp = add_otptoken(master, USER, otptype="hotp")
|
|
|
|
otp1 = hotp.generate(10).decode("ascii")
|
|
otp2 = hotp.generate(11).decode("ascii")
|
|
|
|
master.run_command(
|
|
["ipa", "otptoken-sync", "--user", USER],
|
|
stdin_text=f"{PASSWORD}\n{otp1}\n{otp2}\n",
|
|
)
|
|
otpvalue = hotp.generate(12).decode("ascii")
|
|
kinit_otp(master, USER, password=PASSWORD, otp=otpvalue)
|
|
|
|
otp1 = hotp.generate(20).decode("ascii")
|
|
otp2 = hotp.generate(21).decode("ascii")
|
|
|
|
master.run_command(
|
|
["ipa", "otptoken-sync", otpuid, "--user", USER],
|
|
stdin_text=f"{PASSWORD}\n{otp1}\n{otp2}\n",
|
|
)
|
|
otpvalue = hotp.generate(22).decode("ascii")
|
|
kinit_otp(master, USER, password=PASSWORD, otp=otpvalue)
|
|
|
|
del_otptoken(master, otpuid)
|
|
|
|
def test_2fa_enable_single_prompt(self):
|
|
"""Test ssh with 2FA when single prompt is enabled.
|
|
|
|
Test for : https://pagure.io/SSSD/sssd/issue/3264
|
|
|
|
When [prompting/2fa/sshd] with single_prompt = True is set
|
|
then during ssh it should be prompted with given message
|
|
for first and second factor at once.
|
|
"""
|
|
|
|
master = self.master
|
|
USER1 = 'sshuser1'
|
|
sssd_conf_backup = tasks.FileBackup(master, paths.SSSD_CONF)
|
|
first_prompt = 'Please enter password + OTP token value:'
|
|
add_contents = textwrap.dedent('''
|
|
[prompting/2fa/sshd]
|
|
single_prompt = True
|
|
first_prompt = {0}
|
|
''').format(first_prompt)
|
|
set_sssd_conf(master, add_contents)
|
|
tasks.create_active_user(master, USER1, PASSWORD)
|
|
tasks.kinit_admin(master)
|
|
master.run_command(['ipa', 'user-mod', USER1, '--user-auth-type=otp'])
|
|
try:
|
|
otpuid, totp = add_otptoken(master, USER1, otptype='totp')
|
|
master.run_command(['ipa', 'otptoken-show', otpuid])
|
|
otpvalue = totp.generate(int(time.time())).decode('ascii')
|
|
password = '{0}{1}'.format(PASSWORD, otpvalue)
|
|
tasks.run_ssh_cmd(
|
|
to_host=self.master.external_hostname, username=USER1,
|
|
auth_method="password", password=password
|
|
)
|
|
# check if user listed in output
|
|
cmd = self.master.run_command(['semanage', 'login', '-l'])
|
|
assert USER1 in cmd.stdout_text
|
|
finally:
|
|
master.run_command(['ipa', 'user-del', USER1])
|
|
self.master.run_command(['semanage', 'login', '-D'])
|
|
sssd_conf_backup.restore()
|
|
|
|
def test_2fa_disable_single_prompt(self):
|
|
"""Test ssh with 2FA when single prompt is disabled.
|
|
|
|
Test for : https://pagure.io/SSSD/sssd/issue/3264
|
|
|
|
When [prompting/2fa/sshd] with single_prompt = False is set
|
|
then during ssh it should be prompted with given message
|
|
for first factor and then for second factor.
|
|
|
|
This requires paramiko until the 2-prompt sshpass RFE is
|
|
fulfilled: https://sourceforge.net/p/sshpass/feature-requests/5/
|
|
"""
|
|
if self.master.is_fips_mode:
|
|
pytest.skip("paramiko is not compatible with FIPS mode")
|
|
|
|
master = self.master
|
|
USER2 = 'sshuser2'
|
|
sssd_conf_backup = tasks.FileBackup(master, paths.SSSD_CONF)
|
|
first_prompt = 'Enter first factor:'
|
|
second_prompt = 'Enter second factor:'
|
|
add_contents = textwrap.dedent('''
|
|
[prompting/2fa/sshd]
|
|
single_prompt = False
|
|
first_prompt = {0}
|
|
second_prompt = {1}
|
|
''').format(first_prompt, second_prompt)
|
|
set_sssd_conf(master, add_contents)
|
|
tasks.create_active_user(master, USER2, PASSWORD)
|
|
tasks.kinit_admin(master)
|
|
master.run_command(['ipa', 'user-mod', USER2, '--user-auth-type=otp'])
|
|
try:
|
|
otpuid, totp = add_otptoken(master, USER2, otptype='totp')
|
|
master.run_command(['ipa', 'otptoken-show', otpuid])
|
|
otpvalue = totp.generate(int(time.time())).decode('ascii')
|
|
answers = {
|
|
first_prompt: PASSWORD,
|
|
second_prompt: otpvalue
|
|
}
|
|
ssh_2f(master.hostname, USER2, answers)
|
|
# check if user listed in output
|
|
cmd = self.master.run_command(['semanage', 'login', '-l'])
|
|
assert USER2 in cmd.stdout_text
|
|
finally:
|
|
master.run_command(['ipa', 'user-del', USER2])
|
|
self.master.run_command(['semanage', 'login', '-D'])
|
|
sssd_conf_backup.restore()
|
|
|
|
def test_2fa_only_with_password(self):
|
|
"""Test ssh with 2FA only with the password(first factor) when
|
|
user-auth-type is opt and password.
|
|
|
|
Test for : https://github.com/SSSD/sssd/pull/7500
|
|
|
|
Add the IPA user and user-auth-type set to opt and password.
|
|
Authenticate the user only with password, just press enter
|
|
at `Second factor`
|
|
"""
|
|
master = self.master
|
|
USER3 = 'sshuser3'
|
|
sssd_conf_backup = tasks.FileBackup(master, paths.SSSD_CONF)
|
|
first_prompt = 'Enter first factor:'
|
|
second_prompt = 'Enter second factor:'
|
|
add_contents = textwrap.dedent('''
|
|
[prompting/2fa/sshd]
|
|
single_prompt = False
|
|
first_prompt = {0}
|
|
second_prompt = {1}
|
|
''').format(first_prompt, second_prompt)
|
|
set_sssd_conf(master, add_contents)
|
|
tasks.create_active_user(master, USER3, PASSWORD)
|
|
tasks.kinit_admin(master)
|
|
master.run_command(['ipa', 'user-mod', USER3, '--user-auth-type=otp',
|
|
'--user-auth-type=password'])
|
|
try:
|
|
otpuid, totp = add_otptoken(master, USER3, otptype='totp')
|
|
master.run_command(['ipa', 'otptoken-show', otpuid])
|
|
totp.generate(int(time.time())).decode('ascii')
|
|
otpvalue = "\n"
|
|
tasks.clear_sssd_cache(self.master)
|
|
github_ticket = "https://github.com/SSSD/sssd/pull/7500"
|
|
sssd_version = tasks.get_sssd_version(master)
|
|
rhel_fail = (
|
|
osinfo.id == 'rhel'
|
|
and sssd_version < tasks.parse_version("2.9.5")
|
|
)
|
|
fedora_fail = (
|
|
osinfo.id == 'fedora'
|
|
and sssd_version == tasks.parse_version("2.9.5")
|
|
)
|
|
with xfail_context(rhel_fail or fedora_fail, reason=github_ticket):
|
|
result = ssh_2fa_with_cmd(master,
|
|
self.master.hostname,
|
|
USER3, PASSWORD, otpvalue=otpvalue,
|
|
command="klist")
|
|
print(result.stdout_text)
|
|
assert ('Authentication successful') in result.stdout_text
|
|
assert USER3 in result.stdout_text
|
|
assert (f'Default principal: '
|
|
f'{USER3}@{self.master.domain.realm}' in
|
|
result.stdout_text)
|
|
cmd = self.master.run_command(['semanage', 'login', '-l'])
|
|
assert USER3 in cmd.stdout_text
|
|
finally:
|
|
master.run_command(['ipa', 'user-del', USER3])
|
|
self.master.run_command(['semanage', 'login', '-D'])
|
|
sssd_conf_backup.restore()
|
|
|
|
def test_2fa_with_otp_password(self):
|
|
"""Test ssh with 2FA only with password and otpvalue when
|
|
user-auth-type is opt and password.
|
|
|
|
Test for : https://github.com/SSSD/sssd/pull/7500
|
|
|
|
Add the IPA user and user-auth-type set to opt and password.
|
|
Authenticate the user only with password and otpvalue.
|
|
"""
|
|
master = self.master
|
|
USER4 = 'sshuser4'
|
|
sssd_conf_backup = tasks.FileBackup(master, paths.SSSD_CONF)
|
|
first_prompt = 'Enter first factor:'
|
|
second_prompt = 'Enter second factor:'
|
|
add_contents = textwrap.dedent('''
|
|
[prompting/2fa/sshd]
|
|
single_prompt = False
|
|
first_prompt = {0}
|
|
second_prompt = {1}
|
|
''').format(first_prompt, second_prompt)
|
|
set_sssd_conf(master, add_contents)
|
|
tasks.create_active_user(master, USER4, PASSWORD)
|
|
tasks.kinit_admin(master)
|
|
|
|
master.run_command(['ipa', 'user-mod', USER4, '--user-auth-type=otp',
|
|
'--user-auth-type=password'])
|
|
try:
|
|
otpuid, totp = add_otptoken(master, USER4, otptype='totp')
|
|
master.run_command(['ipa', 'otptoken-show', otpuid])
|
|
otpvalue = totp.generate(int(time.time())).decode('ascii')
|
|
tasks.clear_sssd_cache(self.master)
|
|
result = ssh_2fa_with_cmd(master,
|
|
self.master.hostname,
|
|
USER4, PASSWORD, otpvalue=otpvalue,
|
|
command="klist")
|
|
print(result.stdout_text)
|
|
cmd = self.master.run_command(['semanage', 'login', '-l'])
|
|
# check the output
|
|
assert ('Authentication successful') in result.stdout_text
|
|
assert USER4 in result.stdout_text
|
|
assert (f'Default principal: {USER4}@'
|
|
f'{self.master.domain.realm}' in result.stdout_text)
|
|
assert USER4 in cmd.stdout_text
|
|
finally:
|
|
master.run_command(['ipa', 'user-del', USER4])
|
|
self.master.run_command(['semanage', 'login', '-D'])
|
|
sssd_conf_backup.restore()
|
|
|
|
@pytest.fixture
|
|
def setup_otp_nsslapd(self):
|
|
check_services = self.master.run_command(
|
|
['systemctl', 'list-units', '--state=failed']
|
|
)
|
|
assert "ipa-otpd" not in check_services.stdout_text
|
|
# Be sure no services are running and failed units
|
|
self.master.run_command(['killall', 'ipa-otpd'], raiseonerr=False)
|
|
# setting nsslapd-idletimeout
|
|
new_limit = 30
|
|
conn = self.master.ldap_connect()
|
|
dn = DN(('cn', 'config'))
|
|
entry = conn.get_entry(dn)
|
|
orig_limit = entry.single_value.get('nsslapd-idletimeout')
|
|
ldap_query = textwrap.dedent("""
|
|
dn: cn=config
|
|
changetype: modify
|
|
replace: nsslapd-idletimeout
|
|
nsslapd-idletimeout: {limit}
|
|
""")
|
|
tasks.ldapmodify_dm(self.master, ldap_query.format(limit=new_limit))
|
|
yield
|
|
# cleanup
|
|
tasks.ldapmodify_dm(self.master, ldap_query.format(limit=orig_limit))
|
|
|
|
def test_check_otpd_after_idle_timeout(self, setup_otp_nsslapd):
|
|
"""Test for OTP when the LDAP connection timed out.
|
|
|
|
Test for : https://pagure.io/freeipa/issue/6587
|
|
|
|
ipa-otpd was exiting with failure when LDAP connection timed out.
|
|
Test to verify that when the nsslapd-idletimeout is exceeded (30s idle,
|
|
60s sleep) then the ipa-otpd process should exit without error.
|
|
"""
|
|
since = time.strftime('%Y-%m-%d %H:%M:%S')
|
|
tasks.kinit_admin(self.master)
|
|
otpuid, totp = add_otptoken(self.master, USER, otptype="totp")
|
|
try:
|
|
# kinit with OTP auth
|
|
otpvalue = totp.generate(int(time.time())).decode("ascii")
|
|
kinit_otp(self.master, USER, password=PASSWORD, otp=otpvalue)
|
|
time.sleep(60)
|
|
# ldapsearch will wake up slapd and force walking through
|
|
# the connection list, in order to spot the idle connections
|
|
tasks.ldapsearch_dm(self.master, "", ldap_args=[], scope="base")
|
|
|
|
def test_cb(cmd_jornalctl):
|
|
# check if LDAP connection is timed out
|
|
expected_msg = "Can't contact LDAP server"
|
|
return expected_msg in cmd_jornalctl
|
|
|
|
# ipa-otpd don't flush its logs to syslog immediately
|
|
cmd = ['journalctl', '--since={}'.format(since)]
|
|
tasks.run_repeatedly(
|
|
self.master, command=cmd, test=test_cb, timeout=90)
|
|
failed_services = self.master.run_command(
|
|
['systemctl', 'list-units', '--state=failed']
|
|
)
|
|
assert "ipa-otpd" not in failed_services.stdout_text
|
|
finally:
|
|
del_otptoken(self.master, otpuid)
|
|
|
|
def test_totp_ldap(self):
|
|
master = self.master
|
|
basedn = master.domain.basedn
|
|
USER1 = 'user-forced-otp'
|
|
TMP_PASSWORD = 'Secret1234509'
|
|
binddn = DN(f"uid={USER1},cn=users,cn=accounts,{basedn}")
|
|
|
|
tasks.kinit_admin(master)
|
|
master.run_command(['ipa', 'pwpolicy-mod', '--minlife', '0'])
|
|
tasks.user_add(master, USER1, password=TMP_PASSWORD)
|
|
# Enforce use of OTP token for this user
|
|
master.run_command(['ipa', 'user-mod', USER1,
|
|
'--user-auth-type=otp'])
|
|
try:
|
|
# Change initial password through the IPA endpoint
|
|
url = f'https://{master.hostname}/ipa/session/change_password'
|
|
master.run_command(['curl', '-d', f'user={USER1}',
|
|
'-d', f'old_password={TMP_PASSWORD}',
|
|
'-d', f'new_password={PASSWORD}',
|
|
'--referer', f'https://{master.hostname}/ipa',
|
|
url])
|
|
conn = master.ldap_connect()
|
|
# First, attempt authenticating with a password but without LDAP
|
|
# control to enforce OTP presence and without server-side
|
|
# enforcement of the OTP presence check.
|
|
conn.simple_bind(binddn, f"{PASSWORD}")
|
|
# Next, enforce Password+OTP for a user with OTP token
|
|
master.run_command(['ipa', 'config-mod', '--addattr',
|
|
'ipaconfigstring=EnforceLDAPOTP'])
|
|
# Try to bind without OTP because there is no OTP token yet,
|
|
# the operation should succeed because OTP enforcement is implicit
|
|
# and there is no token yet, so it is allowed.
|
|
conn.simple_bind(binddn, f"{PASSWORD}")
|
|
conn.unbind()
|
|
# Add an OTP token now
|
|
otpuid, totp = add_otptoken(master, USER1, otptype="totp")
|
|
# Next, authenticate with Password+OTP and with the LDAP control
|
|
# this operation should succeed
|
|
otpvalue = totp.generate(int(time.time())).decode("ascii")
|
|
conn = master.ldap_connect()
|
|
conn.simple_bind(binddn, f"{PASSWORD}{otpvalue}",
|
|
client_controls=[
|
|
BooleanControl(
|
|
controlType="2.16.840.1.113730.3.8.10.7",
|
|
booleanValue=True)])
|
|
conn.unbind()
|
|
# Sleep to make sure we are going to use a different token value
|
|
time.sleep(45)
|
|
# Use OTP token again, without LDAP control, should succeed
|
|
# because OTP enforcement is implicit
|
|
otpvalue = totp.generate(int(time.time())).decode("ascii")
|
|
conn = master.ldap_connect()
|
|
conn.simple_bind(binddn, f"{PASSWORD}{otpvalue}")
|
|
conn.unbind()
|
|
# Now, try to authenticate without otp and without control
|
|
# this operation should fail because we have OTP token associated
|
|
# with the user account
|
|
try:
|
|
conn = master.ldap_connect()
|
|
conn.simple_bind(binddn, f"{PASSWORD}")
|
|
conn.unbind()
|
|
except errors.ACIError:
|
|
pass
|
|
# Sleep to make sure we are going to use a different token value
|
|
time.sleep(45)
|
|
# Use OTP token again, without LDAP control, should succeed
|
|
# because OTP enforcement is implicit
|
|
otpvalue = totp.generate(int(time.time())).decode("ascii")
|
|
# Finally, change password again, now that otp is present
|
|
master.run_command(['curl', '-d', f'user={USER1}',
|
|
'-d', f'old_password={PASSWORD}',
|
|
'-d', f'new_password={TMP_PASSWORD}0',
|
|
'-d', f'otp={otpvalue}',
|
|
'--referer', f'https://{master.hostname}/ipa',
|
|
url])
|
|
# Remove token
|
|
del_otptoken(self.master, otpuid)
|
|
master.run_command(['ipa', 'config-mod', '--delattr',
|
|
'ipaconfigstring=EnforceLDAPOTP'])
|
|
finally:
|
|
master.run_command(['ipa', 'pwpolicy-mod', '--minlife', '1'])
|
|
master.run_command(['ipa', 'user-del', USER1])
|
|
|
|
def test_totp_expired_ldap(self):
|
|
master = self.master
|
|
basedn = master.domain.basedn
|
|
USER1 = 'user-expired-otp'
|
|
TMP_PASSWORD = 'Secret1234509'
|
|
binddn = DN(f"uid={USER1},cn=users,cn=accounts,{basedn}")
|
|
controls = [
|
|
BooleanControl(
|
|
controlType="2.16.840.1.113730.3.8.10.7",
|
|
booleanValue=True)
|
|
]
|
|
|
|
tasks.kinit_admin(master)
|
|
master.run_command(['ipa', 'pwpolicy-mod', '--minlife', '0'])
|
|
tasks.user_add(master, USER1, password=TMP_PASSWORD)
|
|
# Enforce use of OTP token for this user
|
|
master.run_command(['ipa', 'user-mod', USER1,
|
|
'--user-auth-type=otp'])
|
|
try:
|
|
# Change initial password through the IPA endpoint
|
|
url = f'https://{master.hostname}/ipa/session/change_password'
|
|
master.run_command(['curl', '-d', f'user={USER1}',
|
|
'-d', f'old_password={TMP_PASSWORD}',
|
|
'-d', f'new_password={PASSWORD}',
|
|
'--referer', f'https://{master.hostname}/ipa',
|
|
url])
|
|
conn = master.ldap_connect()
|
|
# First, attempt authenticating with a password but without LDAP
|
|
# control to enforce OTP presence and without server-side
|
|
# enforcement of the OTP presence check.
|
|
conn.simple_bind(binddn, f"{PASSWORD}")
|
|
|
|
# Add an OTP token and then modify it to be expired
|
|
otpuid, totp = add_otptoken(master, USER1, otptype="totp")
|
|
|
|
# Make sure OTP auth is working
|
|
otpvalue = totp.generate(int(time.time())).decode("ascii")
|
|
conn = master.ldap_connect()
|
|
conn.simple_bind(binddn, f"{PASSWORD}{otpvalue}",
|
|
client_controls=controls)
|
|
conn.unbind()
|
|
|
|
# Modfy token so that is now expired
|
|
args = [
|
|
"ipa",
|
|
"otptoken-mod",
|
|
otpuid,
|
|
"--not-after",
|
|
"20241001010000Z",
|
|
]
|
|
master.run_command(args)
|
|
|
|
# Next, authenticate with Password+OTP again and with the LDAP
|
|
# control this operation should now fail
|
|
time.sleep(45)
|
|
otpvalue = totp.generate(int(time.time())).decode("ascii")
|
|
|
|
conn = master.ldap_connect()
|
|
with pytest.raises(errors.ACIError):
|
|
conn.simple_bind(binddn, f"{PASSWORD}{otpvalue}",
|
|
client_controls=controls)
|
|
|
|
# Sleep to make sure we are going to use a different token value
|
|
time.sleep(45)
|
|
|
|
# Use OTP token again but authenticate over ssh and make sure it
|
|
# doesn't fallthrough to asking for a password
|
|
otpvalue = totp.generate(int(time.time())).decode("ascii")
|
|
answers = {
|
|
'Enter first factor:': PASSWORD,
|
|
'Enter second factor:': otpvalue
|
|
}
|
|
with pytest.raises(AuthenticationException):
|
|
# ssh should fail and NOT ask for a password
|
|
ssh_2f(master.hostname, USER1, answers,
|
|
unwanted_prompt="Password:")
|
|
|
|
# Remove token
|
|
del_otptoken(self.master, otpuid)
|
|
|
|
finally:
|
|
master.run_command(['ipa', 'pwpolicy-mod', '--minlife', '1'])
|
|
master.run_command(['ipa', 'user-del', USER1])
|