trustchain-oc1-orchestral/authentik
Archived
10
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

Merge pull request #134 from BeryJu/consent-mode

Browse source 
Add different Modes to Consent Stage
This commit is contained in:
Jens L 2020-07-20 19:14:33 +02:00 committed by GitHub
parent 88029a4335 d33f0fb2cf
commit 0a196608c7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
17 changed files with 331 additions and 76 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
passbook/flows/planner.py
View file

@ -52,7 +52,8 @@ class FlowPlan:
stage = self.stages[0]
marker = self.markers[0]
LOGGER.debug("f(plan_inst): stage has marker", stage=stage, marker=marker)
if marker.__class__ is not StageMarker:
LOGGER.debug("f(plan_inst): stage has marker", stage=stage, marker=marker)
marked_stage = marker.process(self, stage)
if not marked_stage:
LOGGER.debug("f(plan_inst): marker returned none, next stage", stage=stage)

38
passbook/lib/utils/time.py Normal file
View file

@ -0,0 +1,38 @@
"""Time utilities"""
import datetime
from django.core.exceptions import ValidationError
from django.utils.translation import gettext_lazy as _
ALLOWED_KEYS = (
"days",
"seconds",
"microseconds",
"milliseconds",
"minutes",
"hours",
"weeks",
)
def timedelta_string_validator(value: str):
"""Validator for Django that checks if value can be parsed with `timedelta_from_string`"""
try:
timedelta_from_string(value)
except ValueError as exc:
raise ValidationError(
_("%(value)s is not in the correct format of 'hours=3;minutes=1'."),
params={"value": value},
) from exc
def timedelta_from_string(expr: str) -> datetime.timedelta:
"""Convert a string with the format of 'hours=1;minute=3;seconds=5' to a
`datetime.timedelta` Object with hours = 1, minutes = 3, seconds = 5"""
kwargs = {}
for duration_pair in expr.split(";"):
key, value = duration_pair.split("=")
if key.lower() not in ALLOWED_KEYS:
continue
kwargs[key.lower()] = float(value)
return datetime.timedelta(**kwargs)

14
passbook/providers/saml/migrations/0001_initial.py
View file

@ -3,7 +3,7 @@
import django.db.models.deletion
from django.db import migrations, models
import passbook.providers.saml.utils.time
import passbook.lib.utils.time
class Migration(migrations.Migration):
@ -66,9 +66,7 @@ class Migration(migrations.Migration):
models.TextField(
default="minutes=-5",
help_text="Assertion valid not before current time + this value (Format: hours=-1;minutes=-2;seconds=-3).",
validators=[
passbook.providers.saml.utils.time.timedelta_string_validator
],
validators=[passbook.lib.utils.time.timedelta_string_validator],
),
),
(
@ -76,9 +74,7 @@ class Migration(migrations.Migration):
models.TextField(
default="minutes=5",
help_text="Assertion not valid on or after current time + this value (Format: hours=1;minutes=2;seconds=3).",
validators=[
passbook.providers.saml.utils.time.timedelta_string_validator
],
validators=[passbook.lib.utils.time.timedelta_string_validator],
),
),
(
@ -86,9 +82,7 @@ class Migration(migrations.Migration):
models.TextField(
default="minutes=86400",
help_text="Session not valid on or after current time + this value (Format: hours=1;minutes=2;seconds=3).",
validators=[
passbook.providers.saml.utils.time.timedelta_string_validator
],
validators=[passbook.lib.utils.time.timedelta_string_validator],
),
),
(

2
passbook/providers/saml/models.py
View file

@ -11,7 +11,7 @@ from structlog import get_logger
from passbook.core.models import PropertyMapping, Provider
from passbook.crypto.models import CertificateKeyPair
from passbook.lib.utils.template import render_to_string
from passbook.providers.saml.utils.time import timedelta_string_validator
from passbook.lib.utils.time import timedelta_string_validator
LOGGER = get_logger()

3
passbook/providers/saml/processors/assertion.py
View file

@ -9,10 +9,11 @@ from signxml import XMLSigner, XMLVerifier
from structlog import get_logger
from passbook.core.exceptions import PropertyMappingExpressionException
from passbook.lib.utils.time import timedelta_from_string
from passbook.providers.saml.models import SAMLPropertyMapping, SAMLProvider
from passbook.providers.saml.processors.request_parser import AuthNRequest
from passbook.providers.saml.utils import get_random_id
from passbook.providers.saml.utils.time import get_time_string, timedelta_from_string
from passbook.providers.saml.utils.time import get_time_string
from passbook.sources.saml.exceptions import UnsupportedNameIDFormat
from passbook.sources.saml.processors.constants import (
NS_MAP,

5
passbook/providers/saml/tests/test_utils_time.py
View file

@ -4,10 +4,7 @@ from datetime import timedelta
from django.core.exceptions import ValidationError
from django.test import TestCase
from passbook.providers.saml.utils.time import (
timedelta_from_string,
timedelta_string_validator,
)
from passbook.lib.utils.time import timedelta_from_string, timedelta_string_validator
class TestTimeUtils(TestCase):

36
passbook/providers/saml/utils/time.py
View file

@ -2,42 +2,6 @@
import datetime
from typing import Optional
from django.core.exceptions import ValidationError
from django.utils.translation import gettext_lazy as _
ALLOWED_KEYS = (
"days",
"seconds",
"microseconds",
"milliseconds",
"minutes",
"hours",
"weeks",
)
def timedelta_string_validator(value: str):
"""Validator for Django that checks if value can be parsed with `timedelta_from_string`"""
try:
timedelta_from_string(value)
except ValueError as exc:
raise ValidationError(
_("%(value)s is not in the correct format of 'hours=3;minutes=1'."),
params={"value": value},
) from exc
def timedelta_from_string(expr: str) -> datetime.timedelta:
"""Convert a string with the format of 'hours=1;minute=3;seconds=5' to a
`datetime.timedelta` Object with hours = 1, minutes = 3, seconds = 5"""
kwargs = {}
for duration_pair in expr.split(";"):
key, value = duration_pair.split("=")
if key.lower() not in ALLOWED_KEYS:
continue
kwargs[key.lower()] = float(value)
return datetime.timedelta(**kwargs)
def get_time_string(delta: Optional[datetime.timedelta] = None) -> str:
"""Get Data formatted in SAML format"""

6
passbook/sources/saml/migrations/0003_auto_20200624_1957.py
View file

@ -3,7 +3,7 @@
import django.db.models.deletion
from django.db import migrations, models
import passbook.providers.saml.utils.time
import passbook.lib.utils.time
class Migration(migrations.Migration):
@ -27,9 +27,7 @@ class Migration(migrations.Migration):
field=models.TextField(
default="days=1",
help_text="Time offset when temporary users should be deleted. This only applies if your IDP uses the NameID Format 'transient', and the user doesn't log out manually. (Format: hours=1;minutes=2;seconds=3).",
validators=[
passbook.providers.saml.utils.time.timedelta_string_validator
],
validators=[passbook.lib.utils.time.timedelta_string_validator],
verbose_name="Delete temporary users after",
),
),

2
passbook/sources/saml/models.py
View file

@ -11,7 +11,7 @@ from django.utils.translation import gettext_lazy as _
from passbook.core.models import Source
from passbook.core.types import UILoginButton
from passbook.crypto.models import CertificateKeyPair
from passbook.providers.saml.utils.time import timedelta_string_validator
from passbook.lib.utils.time import timedelta_string_validator
from passbook.sources.saml.processors.constants import (
SAML_NAME_ID_FORMAT_EMAIL,
SAML_NAME_ID_FORMAT_PERSISTENT,

2
passbook/sources/saml/tasks.py
View file

@ -3,7 +3,7 @@ from django.utils.timezone import now
from structlog import get_logger
from passbook.core.models import User
from passbook.providers.saml.utils.time import timedelta_from_string
from passbook.lib.utils.time import timedelta_from_string
from passbook.root.celery import CELERY_APP
from passbook.sources.saml.models import SAMLSource

2
passbook/stages/consent/api.py
View file

@ -11,7 +11,7 @@ class ConsentStageSerializer(ModelSerializer):
class Meta:
model = ConsentStage
fields = ["pk", "name"]
fields = ["pk", "name", "mode", "consent_expire_in"]
class ConsentStageViewSet(ModelViewSet):

2
passbook/stages/consent/forms.py
View file

@ -14,7 +14,7 @@ class ConsentStageForm(forms.ModelForm):
class Meta:
model = ConsentStage
fields = ["name"]
fields = ["name", "mode", "consent_expire_in"]
widgets = {
"name": forms.TextInput(),
}

83
passbook/stages/consent/migrations/0002_auto_20200720_0941.py Normal file
View file

@ -0,0 +1,83 @@
# Generated by Django 3.0.8 on 2020-07-20 09:41
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
import passbook.core.models
import passbook.lib.utils.time
class Migration(migrations.Migration):
dependencies = [
("passbook_core", "0006_auto_20200709_1608"),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
("passbook_stages_consent", "0001_initial"),
]
operations = [
migrations.AddField(
model_name="consentstage",
name="consent_expire_in",
field=models.TextField(
default="weeks=4",
help_text="Offset after which consent expires. (Format: hours=1;minutes=2;seconds=3).",
validators=[passbook.lib.utils.time.timedelta_string_validator],
verbose_name="Consent expires in",
),
),
migrations.AddField(
model_name="consentstage",
name="mode",
field=models.TextField(
choices=[
("always_require", "Always Require"),
("permanent", "Permanent"),
("expiring", "Expiring"),
],
default="always_require",
),
),
migrations.CreateModel(
name="UserConsent",
fields=[
(
"id",
models.AutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
(
"expires",
models.DateTimeField(
default=passbook.core.models.default_token_duration
),
),
("expiring", models.BooleanField(default=True)),
(
"application",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
to="passbook_core.Application",
),
),
(
"user",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="pb_consent",
to=settings.AUTH_USER_MODEL,
),
),
],
options={
"verbose_name": "User Consent",
"verbose_name_plural": "User Consents",
"unique_together": {("user", "application")},
},
),
]

43
passbook/stages/consent/models.py
View file

@ -1,16 +1,42 @@
"""passbook consent stage"""
from typing import Type
from django.db import models
from django.forms import ModelForm
from django.utils.translation import gettext_lazy as _
from django.views import View
from passbook.core.models import Application, ExpiringModel, User
from passbook.flows.models import Stage
from passbook.lib.utils.time import timedelta_string_validator
class ConsentMode(models.TextChoices):
"""Modes a Consent Stage can operate in"""
ALWAYS_REQUIRE = "always_require"
PERMANENT = "permanent"
EXPIRING = "expiring"
class ConsentStage(Stage):
"""Prompt the user for confirmation."""
mode = models.TextField(
choices=ConsentMode.choices, default=ConsentMode.ALWAYS_REQUIRE
)
consent_expire_in = models.TextField(
validators=[timedelta_string_validator],
default="weeks=4",
verbose_name="Consent expires in",
help_text=_(
(
"Offset after which consent expires. "
"(Format: hours=1;minutes=2;seconds=3)."
)
),
)
def type(self) -> Type[View]:
from passbook.stages.consent.stage import ConsentStageView
@ -28,3 +54,20 @@ class ConsentStage(Stage):
verbose_name = _("Consent Stage")
verbose_name_plural = _("Consent Stages")
class UserConsent(ExpiringModel):
"""Consent given by a user for an application"""
# TODO: Remove related_name when oidc provider is v2
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="pb_consent")
application = models.ForeignKey(Application, on_delete=models.CASCADE)
def __str__(self):
return f"User Consent {self.application} by {self.user}"
class Meta:
unique_together = (("user", "application"),)
verbose_name = _("User Consent")
verbose_name_plural = _("User Consents")

42
passbook/stages/consent/stage.py
View file

@ -1,10 +1,15 @@
"""passbook consent stage"""
from typing import Any, Dict, List
from django.http import HttpRequest, HttpResponse
from django.utils.timezone import now
from django.views.generic import FormView
from passbook.flows.planner import PLAN_CONTEXT_APPLICATION
from passbook.flows.stage import StageView
from passbook.lib.utils.time import timedelta_from_string
from passbook.stages.consent.forms import ConsentForm
from passbook.stages.consent.models import ConsentMode, ConsentStage, UserConsent
PLAN_CONTEXT_CONSENT_TEMPLATE = "consent_template"
@ -26,5 +31,40 @@ class ConsentStageView(FormView, StageView):
return [template_name]
return super().get_template_names()
def form_valid(self, form):
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
current_stage: ConsentStage = self.executor.current_stage
# For always require, we always show the form
if current_stage.mode == ConsentMode.ALWAYS_REQUIRE:
return super().get(request, *args, **kwargs)
# at this point we need to check consent from database
if PLAN_CONTEXT_APPLICATION not in self.executor.plan.context:
# No application in this plan, hence we can't check DB and require user consent
return super().get(request, *args, **kwargs)
application = self.executor.plan.context[PLAN_CONTEXT_APPLICATION]
# TODO: Check for user in plan?
if UserConsent.filter_not_expired(
user=self.request.user, application=application
).exists():
return self.executor.stage_ok()
# No consent found, show form
return super().get(request, *args, **kwargs)
def form_valid(self, form: ConsentForm) -> HttpResponse:
current_stage: ConsentStage = self.executor.current_stage
if PLAN_CONTEXT_APPLICATION not in self.executor.plan.context:
return self.executor.stage_ok()
application = self.executor.plan.context[PLAN_CONTEXT_APPLICATION]
# Since we only get here when no consent exists, we can create it without update
if current_stage.mode == ConsentMode.PERMANENT:
UserConsent.objects.create(
user=self.request.user, application=application, expiring=False
)
if current_stage.mode == ConsentMode.EXPIRING:
UserConsent.objects.create(
user=self.request.user,
application=application,
expires=now() + timedelta_from_string(current_stage.consent_expire_in),
)
return self.executor.stage_ok()

112
passbook/stages/consent/tests.py
View file

@ -1,14 +1,17 @@
"""consent tests"""
from time import sleep
from django.shortcuts import reverse
from django.test import Client, TestCase
from django.utils.encoding import force_text
from passbook.core.models import User
from passbook.core.models import Application, User
from passbook.core.tasks import clean_expired_models
from passbook.flows.markers import StageMarker
from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
from passbook.flows.planner import FlowPlan
from passbook.flows.planner import PLAN_CONTEXT_APPLICATION, FlowPlan
from passbook.flows.views import SESSION_KEY_PLAN
from passbook.stages.consent.models import ConsentStage
from passbook.stages.consent.models import ConsentMode, ConsentStage, UserConsent
class TestConsentStage(TestCase):
@ -19,28 +22,29 @@ class TestConsentStage(TestCase):
self.user = User.objects.create_user(
username="unittest", email="test@beryju.org"
)
self.application = Application.objects.create(
name="test-application", slug="test-application",
)
self.client = Client()
self.flow = Flow.objects.create(
def test_always_required(self):
"""Test always required consent"""
flow = Flow.objects.create(
name="test-consent",
slug="test-consent",
designation=FlowDesignation.AUTHENTICATION,
)
self.stage = ConsentStage.objects.create(name="consent",)
FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)
def test_valid(self):
"""Test valid consent"""
plan = FlowPlan(
flow_pk=self.flow.pk.hex, stages=[self.stage], markers=[StageMarker()]
stage = ConsentStage.objects.create(
name="consent", mode=ConsentMode.ALWAYS_REQUIRE
)
FlowStageBinding.objects.create(target=flow, stage=stage, order=2)
plan = FlowPlan(flow_pk=flow.pk.hex, stages=[stage], markers=[StageMarker()])
session = self.client.session
session[SESSION_KEY_PLAN] = plan
session.save()
response = self.client.post(
reverse(
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
),
reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}),
{},
)
self.assertEqual(response.status_code, 200)
@ -48,3 +52,83 @@ class TestConsentStage(TestCase):
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
self.assertFalse(UserConsent.objects.filter(user=self.user).exists())
def test_permanent(self):
"""Test permanent consent from user"""
self.client.force_login(self.user)
flow = Flow.objects.create(
name="test-consent",
slug="test-consent",
designation=FlowDesignation.AUTHENTICATION,
)
stage = ConsentStage.objects.create(name="consent", mode=ConsentMode.PERMANENT)
FlowStageBinding.objects.create(target=flow, stage=stage, order=2)
plan = FlowPlan(
flow_pk=flow.pk.hex,
stages=[stage],
markers=[StageMarker()],
context={PLAN_CONTEXT_APPLICATION: self.application},
)
session = self.client.session
session[SESSION_KEY_PLAN] = plan
session.save()
response = self.client.post(
reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}),
{},
)
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
self.assertTrue(
UserConsent.objects.filter(
user=self.user, application=self.application
).exists()
)
def test_expire(self):
"""Test expiring consent from user"""
self.client.force_login(self.user)
flow = Flow.objects.create(
name="test-consent",
slug="test-consent",
designation=FlowDesignation.AUTHENTICATION,
)
stage = ConsentStage.objects.create(
name="consent", mode=ConsentMode.EXPIRING, consent_expire_in="seconds=1"
)
FlowStageBinding.objects.create(target=flow, stage=stage, order=2)
plan = FlowPlan(
flow_pk=flow.pk.hex,
stages=[stage],
markers=[StageMarker()],
context={PLAN_CONTEXT_APPLICATION: self.application},
)
session = self.client.session
session[SESSION_KEY_PLAN] = plan
session.save()
response = self.client.post(
reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}),
{},
)
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
self.assertTrue(
UserConsent.objects.filter(
user=self.user, application=self.application
).exists()
)
sleep(1)
clean_expired_models()
self.assertFalse(
UserConsent.objects.filter(
user=self.user, application=self.application
).exists()
)

12
swagger.yaml
View file

@ -6609,6 +6609,18 @@ definitions:
title: Name
type: string
minLength: 1
mode:
title: Mode
type: string
enum:
- always_require
- permanent
- expiring
consent_expire_in:
title: Consent expires in
description: 'Offset after which consent expires. (Format: hours=1;minutes=2;seconds=3).'
type: string
minLength: 1
DummyStage:
required:
- name