e2e: add tests for TOTP Setup, static OTP Setup and otp validation
This commit is contained in:
parent
f294791d41
commit
769ce1c642
|
@ -0,0 +1,141 @@
|
||||||
|
"""test flow with otp stages"""
|
||||||
|
from base64 import b32decode
|
||||||
|
from sys import platform
|
||||||
|
from time import sleep
|
||||||
|
from unittest.case import skipUnless
|
||||||
|
from urllib.parse import parse_qs, urlparse
|
||||||
|
|
||||||
|
from django_otp.oath import TOTP
|
||||||
|
from django_otp.plugins.otp_static.models import StaticDevice, StaticToken
|
||||||
|
from django_otp.plugins.otp_totp.models import TOTPDevice
|
||||||
|
from selenium.webdriver.common.by import By
|
||||||
|
from selenium.webdriver.common.keys import Keys
|
||||||
|
|
||||||
|
from e2e.utils import USER, SeleniumTestCase
|
||||||
|
from passbook.flows.models import Flow, FlowStageBinding
|
||||||
|
from passbook.stages.otp_validate.models import OTPValidateStage
|
||||||
|
|
||||||
|
|
||||||
|
@skipUnless(platform.startswith("linux"), "requires local docker")
|
||||||
|
class TestFlowsOTP(SeleniumTestCase):
|
||||||
|
"""test flow with otp stages"""
|
||||||
|
|
||||||
|
def test_otp_validate(self):
|
||||||
|
"""test flow with otp stages"""
|
||||||
|
sleep(1)
|
||||||
|
# Setup TOTP Device
|
||||||
|
user = USER()
|
||||||
|
device = TOTPDevice.objects.create(user=user, confirmed=True, digits=6)
|
||||||
|
|
||||||
|
flow: Flow = Flow.objects.get(slug="default-authentication-flow")
|
||||||
|
# Move the user_login stage to order 3
|
||||||
|
FlowStageBinding.objects.filter(target=flow, order=2).update(order=3)
|
||||||
|
FlowStageBinding.objects.create(
|
||||||
|
target=flow, order=2, stage=OTPValidateStage.objects.create()
|
||||||
|
)
|
||||||
|
|
||||||
|
self.driver.get(f"{self.live_server_url}/flows/{flow.slug}/")
|
||||||
|
self.driver.find_element(By.ID, "id_uid_field").click()
|
||||||
|
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
|
||||||
|
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
|
||||||
|
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
|
||||||
|
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
|
||||||
|
|
||||||
|
# Get expected token
|
||||||
|
totp = TOTP(device.bin_key, device.step, device.t0, device.digits, device.drift)
|
||||||
|
self.driver.find_element(By.ID, "id_code").send_keys(totp.token())
|
||||||
|
self.driver.find_element(By.ID, "id_code").send_keys(Keys.ENTER)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(By.XPATH, "//a[contains(@href, '/-/user/')]").text,
|
||||||
|
USER().username,
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_otp_totp_setup(self):
|
||||||
|
"""test TOTP Setup stage"""
|
||||||
|
flow: Flow = Flow.objects.get(slug="default-authentication-flow")
|
||||||
|
|
||||||
|
self.driver.get(f"{self.live_server_url}/flows/{flow.slug}/")
|
||||||
|
self.driver.find_element(By.ID, "id_uid_field").click()
|
||||||
|
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
|
||||||
|
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
|
||||||
|
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
|
||||||
|
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(By.XPATH, "//a[contains(@href, '/-/user/')]").text,
|
||||||
|
USER().username,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-page__header").click()
|
||||||
|
self.driver.find_element(By.XPATH, "//a[contains(@href, '/-/user/')]").click()
|
||||||
|
self.wait_for_url(self.url("passbook_core:user-settings"))
|
||||||
|
|
||||||
|
self.driver.find_element(By.LINK_TEXT, "Time-based OTP").click()
|
||||||
|
|
||||||
|
# Remember the current URL as we should end up back here
|
||||||
|
destination_url = self.driver.current_url
|
||||||
|
|
||||||
|
self.driver.find_element(
|
||||||
|
By.CSS_SELECTOR, ".pf-c-card__body a.pf-c-button"
|
||||||
|
).click()
|
||||||
|
|
||||||
|
otp_uri = self.driver.find_element(
|
||||||
|
By.CSS_SELECTOR, "#flow-body > div > form > div:nth-child(3) > div"
|
||||||
|
).get_attribute("aria-label")
|
||||||
|
|
||||||
|
# Parse the OTP URI, extract the secret and get the next token
|
||||||
|
otp_args = urlparse(otp_uri)
|
||||||
|
self.assertEqual(otp_args.scheme, "otpauth")
|
||||||
|
otp_qs = parse_qs(otp_args.query)
|
||||||
|
secret_key = b32decode(otp_qs["secret"][0])
|
||||||
|
|
||||||
|
totp = TOTP(secret_key)
|
||||||
|
|
||||||
|
self.driver.find_element(By.ID, "id_code").send_keys(totp.token())
|
||||||
|
self.driver.find_element(By.ID, "id_code").send_keys(Keys.ENTER)
|
||||||
|
|
||||||
|
self.wait_for_url(destination_url)
|
||||||
|
sleep(1)
|
||||||
|
|
||||||
|
self.assertTrue(TOTPDevice.objects.filter(user=USER(), confirmed=True).exists())
|
||||||
|
|
||||||
|
def test_otp_static_setup(self):
|
||||||
|
"""test Static OTP Setup stage"""
|
||||||
|
flow: Flow = Flow.objects.get(slug="default-authentication-flow")
|
||||||
|
|
||||||
|
self.driver.get(f"{self.live_server_url}/flows/{flow.slug}/")
|
||||||
|
self.driver.find_element(By.ID, "id_uid_field").click()
|
||||||
|
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
|
||||||
|
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
|
||||||
|
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
|
||||||
|
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(By.XPATH, "//a[contains(@href, '/-/user/')]").text,
|
||||||
|
USER().username,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-page__header").click()
|
||||||
|
self.driver.find_element(By.XPATH, "//a[contains(@href, '/-/user/')]").click()
|
||||||
|
self.wait_for_url(self.url("passbook_core:user-settings"))
|
||||||
|
|
||||||
|
self.driver.find_element(By.LINK_TEXT, "Static OTP").click()
|
||||||
|
|
||||||
|
# Remember the current URL as we should end up back here
|
||||||
|
destination_url = self.driver.current_url
|
||||||
|
|
||||||
|
self.driver.find_element(
|
||||||
|
By.CSS_SELECTOR, ".pf-c-card__body a.pf-c-button"
|
||||||
|
).click()
|
||||||
|
token = self.driver.find_element(
|
||||||
|
By.CSS_SELECTOR, ".pb-otp-tokens li:nth-child(1)"
|
||||||
|
).text
|
||||||
|
|
||||||
|
self.driver.find_element(By.CSS_SELECTOR, "button[type=submit]").click()
|
||||||
|
|
||||||
|
self.wait_for_url(destination_url)
|
||||||
|
sleep(1)
|
||||||
|
|
||||||
|
self.assertTrue(
|
||||||
|
StaticDevice.objects.filter(user=USER(), confirmed=True).exists()
|
||||||
|
)
|
||||||
|
device = StaticDevice.objects.filter(user=USER(), confirmed=True).first()
|
||||||
|
self.assertTrue(StaticToken.objects.filter(token=token, device=device).exists())
|
|
@ -20,7 +20,7 @@ class TestFlowsStageSetup(SeleniumTestCase):
|
||||||
"""test password change flow"""
|
"""test password change flow"""
|
||||||
# Ensure that password stage has change_flow set
|
# Ensure that password stage has change_flow set
|
||||||
flow = Flow.objects.get(
|
flow = Flow.objects.get(
|
||||||
slug="default-password-change", designation=FlowDesignation.STAGE_SETUP,
|
slug="default-password-change", designation=FlowDesignation.STAGE_CONFIGURATION,
|
||||||
)
|
)
|
||||||
|
|
||||||
stages = PasswordStage.objects.filter(name="default-authentication-password")
|
stages = PasswordStage.objects.filter(name="default-authentication-password")
|
||||||
|
|
|
@ -1,5 +1,6 @@
|
||||||
"""test OAuth Provider flow"""
|
"""test OAuth Provider flow"""
|
||||||
from sys import platform
|
from sys import platform
|
||||||
|
from time import sleep
|
||||||
from typing import Any, Dict, Optional
|
from typing import Any, Dict, Optional
|
||||||
from unittest.case import skipUnless
|
from unittest.case import skipUnless
|
||||||
|
|
||||||
|
@ -159,6 +160,8 @@ class TestProviderOAuth2Github(SeleniumTestCase):
|
||||||
),
|
),
|
||||||
).click()
|
).click()
|
||||||
|
|
||||||
|
sleep(1)
|
||||||
|
|
||||||
self.wait_for_url("http://localhost:3000/?orgId=1")
|
self.wait_for_url("http://localhost:3000/?orgId=1")
|
||||||
self.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
|
self.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
|
|
|
@ -1,5 +1,4 @@
|
||||||
"""passbook e2e testing utilities"""
|
"""passbook e2e testing utilities"""
|
||||||
from functools import lru_cache
|
|
||||||
from glob import glob
|
from glob import glob
|
||||||
from importlib.util import module_from_spec, spec_from_file_location
|
from importlib.util import module_from_spec, spec_from_file_location
|
||||||
from inspect import getmembers, isfunction
|
from inspect import getmembers, isfunction
|
||||||
|
@ -23,7 +22,6 @@ from structlog import get_logger
|
||||||
from passbook.core.models import User
|
from passbook.core.models import User
|
||||||
|
|
||||||
|
|
||||||
@lru_cache
|
|
||||||
# pylint: disable=invalid-name
|
# pylint: disable=invalid-name
|
||||||
def USER() -> User: # noqa
|
def USER() -> User: # noqa
|
||||||
"""Cached function that always returns pbadmin"""
|
"""Cached function that always returns pbadmin"""
|
||||||
|
|
|
@ -7392,6 +7392,13 @@ definitions:
|
||||||
title: Name
|
title: Name
|
||||||
type: string
|
type: string
|
||||||
minLength: 1
|
minLength: 1
|
||||||
|
configure_flow:
|
||||||
|
title: Configure flow
|
||||||
|
description: Flow used by an authenticated user to configure this Stage. If
|
||||||
|
empty, user will not be able to configure this stage.
|
||||||
|
type: string
|
||||||
|
format: uuid
|
||||||
|
x-nullable: true
|
||||||
token_count:
|
token_count:
|
||||||
title: Token count
|
title: Token count
|
||||||
type: integer
|
type: integer
|
||||||
|
|
Reference in New Issue