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

WIP Use Flows for Sources and Providers (#32)

Browse source 
* core: start migrating to flows for authorisation

* sources/oauth: start type-hinting

* core: create default user

* core: only show user delete button if an unenrollment flow exists

* flows: Correctly check initial policies on flow with context

* policies: add more verbosity to engine

* sources/oauth: migrate to flows

* sources/oauth: fix typing errors

* flows: add more tests

* sources/oauth: start implementing unittests

* sources/ldap: add option to disable user sync, move connection init to model

* sources/ldap: re-add default PropertyMappings

* providers/saml: re-add default PropertyMappings

* admin: fix missing stage count

* stages/identification: fix sources not being shown

* crypto: fix being unable to save with private key

* crypto: re-add default self-signed keypair

* policies: rewrite cache_key to prevent wrong cache

* sources/saml: migrate to flows for auth and enrollment

* stages/consent: add new stage

* admin: fix PropertyMapping widget not rendering properly

* core: provider.authorization_flow is mandatory

* flows: add support for "autosubmit" attribute on form

* flows: add InMemoryStage for dynamic stages

* flows: optionally allow empty flows from FlowPlanner

* providers/saml: update to authorization_flow

* sources/*: fix flow executor URL

* flows: fix pylint error

* flows: wrap responses in JSON object to easily handle redirects

* flow: dont cache plan's context

* providers/oauth: rewrite OAuth2 Provider to use flows

* providers/*: update docstrings of models

* core: fix forms not passing help_text through safe

* flows: fix HttpResponses not being converted to JSON

* providers/oidc: rewrite to use flows

* flows: fix linting
This commit is contained in:
Jens L 2020-06-07 16:35:08 +02:00 committed by GitHub
parent f91e02a0ec
commit 4915205678
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
81 changed files with 1609 additions and 529 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
passbook/admin/forms/source.py
View file

@ -1,4 +1,17 @@
"""passbook core source form fields"""
SOURCE_FORM_FIELDS = ["name", "slug", "enabled"]
SOURCE_SERIALIZER_FIELDS = ["pk", "name", "slug", "enabled"]
SOURCE_FORM_FIELDS = [
"name",
"slug",
"enabled",
"authentication_flow",
"enrollment_flow",
]
SOURCE_SERIALIZER_FIELDS = [
"pk",
"name",
"slug",
"enabled",
"authentication_flow",
"enrollment_flow",
]

2
passbook/admin/templates/administration/provider/list.html
View file

@ -29,7 +29,7 @@
{% for type, name in types.items %}
<li>
<a class="pf-c-dropdown__menu-item" href="{% url 'passbook_admin:provider-create' %}?type={{ type }}&back={{ request.get_full_path }}">
{{ name|verbose_name }}
{{ name|verbose_name }}<br>
<small>
{{ name|doc }}
</small>

4
passbook/admin/templates/generic/create.html
View file

@ -5,14 +5,14 @@
{% block above_form %}
<h1>
{% blocktrans with type=form|form_verbose_name|title %}
{% blocktrans with type=form|form_verbose_name %}
Create {{ type }}
{% endblocktrans %}
</h1>
{% endblock %}
{% block action %}
{% blocktrans with type=form|form_verbose_name|title %}
{% blocktrans with type=form|form_verbose_name %}
Create {{ type }}
{% endblocktrans %}
{% endblock %}

1
passbook/core/api/applications.py
View file

@ -15,7 +15,6 @@ class ApplicationSerializer(ModelSerializer):
"pk",
"name",
"slug",
"skip_authorization",
"provider",
"meta_launch_url",
"meta_icon_url",

2
passbook/core/api/providers.py
View file

@ -17,7 +17,7 @@ class ProviderSerializer(ModelSerializer):
class Meta:
model = Provider
fields = ["pk", "property_mappings", "__type__"]
fields = ["pk", "authorization_flow", "property_mappings", "__type__"]
class ProviderViewSet(ReadOnlyModelViewSet):

1
passbook/core/forms/applications.py
View file

@ -19,7 +19,6 @@ class ApplicationForm(forms.ModelForm):
fields = [
"name",
"slug",
"skip_authorization",
"provider",
"meta_launch_url",
"meta_icon_url",

52
passbook/core/migrations/0002_auto_20200523_1133.py Normal file
View file

@ -0,0 +1,52 @@
# Generated by Django 3.0.6 on 2020-05-23 11:33
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("passbook_flows", "0003_auto_20200523_1133"),
("passbook_core", "0001_initial"),
]
operations = [
migrations.RemoveField(model_name="application", name="skip_authorization",),
migrations.AddField(
model_name="source",
name="authentication_flow",
field=models.ForeignKey(
blank=True,
default=None,
help_text="Flow to use when authenticating existing users.",
null=True,
on_delete=django.db.models.deletion.SET_NULL,
related_name="source_authentication",
to="passbook_flows.Flow",
),
),
migrations.AddField(
model_name="source",
name="enrollment_flow",
field=models.ForeignKey(
blank=True,
default=None,
help_text="Flow to use when enrolling new users.",
null=True,
on_delete=django.db.models.deletion.SET_NULL,
related_name="source_enrollment",
to="passbook_flows.Flow",
),
),
migrations.AddField(
model_name="provider",
name="authorization_flow",
field=models.ForeignKey(
help_text="Flow used when authorizing this provider.",
on_delete=django.db.models.deletion.CASCADE,
related_name="provider_authorization",
to="passbook_flows.Flow",
),
),
]

4
passbook/core/migrations/0002_default_user.py → passbook/core/migrations/0003_default_user.py
View file

@ -12,7 +12,7 @@ def create_default_user(apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
pbadmin = User.objects.create(
username="pbadmin", email="root@localhost", name="passbook Default Admin"
)
pbadmin.set_password("pbadmin") # nosec
pbadmin.set_password("pbadmin") # noqa # nosec
pbadmin.is_superuser = True
pbadmin.is_staff = True
pbadmin.save()
@ -21,7 +21,7 @@ def create_default_user(apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
class Migration(migrations.Migration):
dependencies = [
("passbook_core", "0001_initial"),
("passbook_core", "0002_auto_20200523_1133"),
]
operations = [

28
passbook/core/models.py
View file

@ -16,6 +16,7 @@ from structlog import get_logger
from passbook.core.exceptions import PropertyMappingExpressionException
from passbook.core.signals import password_changed
from passbook.core.types import UILoginButton, UIUserSettings
from passbook.flows.models import Flow
from passbook.lib.models import CreatedUpdatedModel
from passbook.policies.models import PolicyBindingModel
@ -75,6 +76,13 @@ class User(GuardianUserMixin, AbstractUser):
class Provider(models.Model):
"""Application-independent Provider instance. For example SAML2 Remote, OAuth2 Application"""
authorization_flow = models.ForeignKey(
Flow,
on_delete=models.CASCADE,
help_text=_("Flow used when authorizing this provider."),
related_name="provider_authorization",
)
property_mappings = models.ManyToManyField(
"PropertyMapping", default=None, blank=True
)
@ -95,7 +103,6 @@ class Application(PolicyBindingModel):
name = models.TextField(help_text=_("Application's display Name."))
slug = models.SlugField(help_text=_("Internal application name, used in URLs."))
skip_authorization = models.BooleanField(default=False)
provider = models.OneToOneField(
"Provider", null=True, blank=True, default=None, on_delete=models.SET_DEFAULT
)
@ -128,6 +135,25 @@ class Source(PolicyBindingModel):
"PropertyMapping", default=None, blank=True
)
authentication_flow = models.ForeignKey(
Flow,
blank=True,
null=True,
default=None,
on_delete=models.SET_NULL,
help_text=_("Flow to use when authenticating existing users."),
related_name="source_authentication",
)
enrollment_flow = models.ForeignKey(
Flow,
blank=True,
null=True,
default=None,
on_delete=models.SET_NULL,
help_text=_("Flow to use when enrolling new users."),
related_name="source_enrollment",
)
form = "" # ModelForm-based class ued to create/edit instance
objects = InheritanceManager()

37
passbook/core/templates/login/base.html
View file

@ -20,3 +20,40 @@
</form>
{% endblock %}
</div>
<footer class="pf-c-login__main-footer">
{% if config.login.subtext %}
<p>{{ config.login.subtext }}</p>
{% endif %}
<ul class="pf-c-login__main-footer-links">
{% for source in sources %}
<li class="pf-c-login__main-footer-links-item">
<a href="{{ source.url }}" class="pf-c-login__main-footer-links-item-link">
{% if source.icon_path %}
<img src="{% static source.icon_path %}" alt="{{ source.name }}">
{% elif source.icon_url %}
<img src="icon_url" alt="{{ source.name }}">
{% else %}
<i class="pf-icon pf-icon-arrow" title="{{ source.name }}"></i>
{% endif %}
</a>
</li>
{% endfor %}
</ul>
{% if enroll_url or recovery_url %}
<div class="pf-c-login__main-footer-band">
{% if enroll_url %}
<p class="pf-c-login__main-footer-band-item">
{% trans 'Need an account?' %}
<a href="{{ enroll_url }}">{% trans 'Sign up.' %}</a>
</p>
{% endif %}
{% if recovery_url %}
<p class="pf-c-login__main-footer-band-item">
<a href="{{ recovery_url }}">
{% trans 'Forgot username or password?' %}
</a>
</p>
{% endif %}
</div>
{% endif %}
</footer>

7
passbook/core/templates/partials/form_horizontal.html
View file

@ -28,6 +28,9 @@
</label>
<div class="pf-c-form__horizontal-group">
{{ field|css_class:"pf-c-form-control" }}
{% if field.help_text %}
<p class="pf-c-form__helper-text">{{ field.help_text|safe }}</p>
{% endif %}
</div>
{% elif field.field.widget|fieldtype == 'CheckboxInput' %}
<div class="pf-c-form__horizontal-group">
@ -36,7 +39,7 @@
<label class="pf-c-check__label" for="{{ field.name }}-{{ forloop.counter0 }}">{{ field.label }}</label>
</div>
{% if field.help_text %}
<p class="pf-c-form__helper-text">{{ field.help_text }}</p>
<p class="pf-c-form__helper-text">{{ field.help_text|safe }}</p>
{% endif %}
</div>
{% else %}
@ -49,7 +52,7 @@
<div class="c-form__horizontal-group">
{{ field|css_class:'pf-c-form-control' }}
{% if field.help_text %}
<p class="pf-c-form__helper-text">{{ field.help_text }}</p>
<p class="pf-c-form__helper-text">{{ field.help_text|safe }}</p>
{% endif %}
</div>
{% endif %}

2
passbook/core/views/access.py
View file

@ -35,6 +35,6 @@ class AccessMixin:
def user_has_access(self, application: Application, user: User) -> PolicyResult:
"""Check if user has access to application."""
LOGGER.debug("Checking permissions", user=user, application=application)
policy_engine = PolicyEngine(application.policies.all(), user, self.request)
policy_engine = PolicyEngine(application, user, self.request)
policy_engine.build()
return policy_engine.result

4
passbook/core/views/overview.py
View file

@ -16,9 +16,7 @@ class OverviewView(LoginRequiredMixin, TemplateView):
def get_context_data(self, **kwargs):
kwargs["applications"] = []
for application in Application.objects.all().order_by("name"):
engine = PolicyEngine(
application.policies.all(), self.request.user, self.request
)
engine = PolicyEngine(application, self.request.user, self.request)
engine.build()
if engine.passing:
kwargs["applications"].append(application)

8
passbook/crypto/builder.py
View file

@ -36,11 +36,11 @@ class CertificateBuilder:
x509.Name(
[
x509.NameAttribute(
NameOID.COMMON_NAME, u"passbook Self-signed Certificate",
NameOID.COMMON_NAME, "passbook Self-signed Certificate",
),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, u"passbook"),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "passbook"),
x509.NameAttribute(
NameOID.ORGANIZATIONAL_UNIT_NAME, u"Self-signed"
NameOID.ORGANIZATIONAL_UNIT_NAME, "Self-signed"
),
]
)
@ -49,7 +49,7 @@ class CertificateBuilder:
x509.Name(
[
x509.NameAttribute(
NameOID.COMMON_NAME, u"passbook Self-signed Certificate",
NameOID.COMMON_NAME, "passbook Self-signed Certificate",
),
]
)

29
passbook/flows/migrations/0003_auto_20200523_1133.py Normal file
View file

@ -0,0 +1,29 @@
# Generated by Django 3.0.6 on 2020-05-23 11:33
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("passbook_flows", "0002_default_flows"),
]
operations = [
migrations.AlterField(
model_name="flow",
name="designation",
field=models.CharField(
choices=[
("authentication", "Authentication"),
("authorization", "Authorization"),
("invalidation", "Invalidation"),
("enrollment", "Enrollment"),
("unenrollment", "Unrenollment"),
("recovery", "Recovery"),
("password_change", "Password Change"),
],
max_length=100,
),
),
]

131
passbook/flows/migrations/0004_source_flows.py Normal file
View file

@ -0,0 +1,131 @@
# Generated by Django 3.0.6 on 2020-05-23 15:47
from django.apps.registry import Apps
from django.db import migrations
from django.db.backends.base.schema import BaseDatabaseSchemaEditor
from passbook.flows.models import FlowDesignation
from passbook.stages.prompt.models import FieldTypes
FLOW_POLICY_EXPRESSION = """{{ pb_is_sso_flow }}"""
PROMPT_POLICY_EXPRESSION = """
{% if pb_flow_plan.context.prompt_data.username %}
False
{% else %}
True
{% endif %}
"""
def create_default_source_enrollment_flow(
apps: Apps, schema_editor: BaseDatabaseSchemaEditor
):
Flow = apps.get_model("passbook_flows", "Flow")
FlowStageBinding = apps.get_model("passbook_flows", "FlowStageBinding")
PolicyBinding = apps.get_model("passbook_policies", "PolicyBinding")
ExpressionPolicy = apps.get_model(
"passbook_policies_expression", "ExpressionPolicy"
)
PromptStage = apps.get_model("passbook_stages_prompt", "PromptStage")
Prompt = apps.get_model("passbook_stages_prompt", "Prompt")
UserWriteStage = apps.get_model("passbook_stages_user_write", "UserWriteStage")
UserLoginStage = apps.get_model("passbook_stages_user_login", "UserLoginStage")
db_alias = schema_editor.connection.alias
# Create a policy that only allows this flow when doing an SSO Request
flow_policy = ExpressionPolicy.objects.create(
name="default-source-enrollment-if-sso", expression=FLOW_POLICY_EXPRESSION
)
# This creates a Flow used by sources to enroll users
# It makes sure that a username is set, and if not, prompts the user for a Username
flow = Flow.objects.create(
name="default-source-enrollment",
slug="default-source-enrollment",
designation=FlowDesignation.ENROLLMENT,
)
PolicyBinding.objects.create(policy=flow_policy, target=flow, order=0)
# PromptStage to ask user for their username
prompt_stage = PromptStage.objects.create(
name="default-source-enrollment-username-prompt",
)
prompt_stage.fields.add(
Prompt.objects.create(
field_key="username",
label="Username",
type=FieldTypes.TEXT,
required=True,
placeholder="Username",
)
)
# Policy to only trigger prompt when no username is given
prompt_policy = ExpressionPolicy.objects.create(
name="default-source-enrollment-if-username",
expression=PROMPT_POLICY_EXPRESSION,
)
# UserWrite stage to create the user, and login stage to log user in
user_write = UserWriteStage.objects.create(name="default-source-enrollment-write")
user_login = UserLoginStage.objects.create(name="default-source-enrollment-login")
binding = FlowStageBinding.objects.create(flow=flow, stage=prompt_stage, order=0)
PolicyBinding.objects.create(policy=prompt_policy, target=binding)
FlowStageBinding.objects.create(flow=flow, stage=user_write, order=1)
FlowStageBinding.objects.create(flow=flow, stage=user_login, order=2)
def create_default_source_authentication_flow(
apps: Apps, schema_editor: BaseDatabaseSchemaEditor
):
Flow = apps.get_model("passbook_flows", "Flow")
FlowStageBinding = apps.get_model("passbook_flows", "FlowStageBinding")
PolicyBinding = apps.get_model("passbook_policies", "PolicyBinding")
ExpressionPolicy = apps.get_model(
"passbook_policies_expression", "ExpressionPolicy"
)
UserLoginStage = apps.get_model("passbook_stages_user_login", "UserLoginStage")
db_alias = schema_editor.connection.alias
# Create a policy that only allows this flow when doing an SSO Request
flow_policy = ExpressionPolicy.objects.create(
name="default-source-authentication-if-sso", expression=FLOW_POLICY_EXPRESSION
)
# This creates a Flow used by sources to authenticate users
flow = Flow.objects.create(
name="default-source-authentication",
slug="default-source-authentication",
designation=FlowDesignation.AUTHENTICATION,
)
PolicyBinding.objects.create(policy=flow_policy, target=flow, order=0)
user_login = UserLoginStage.objects.create(
name="default-source-authentication-login"
)
FlowStageBinding.objects.create(flow=flow, stage=user_login, order=0)
class Migration(migrations.Migration):
dependencies = [
("passbook_flows", "0003_auto_20200523_1133"),
("passbook_policies", "0001_initial"),
("passbook_policies_expression", "0001_initial"),
("passbook_stages_prompt", "0001_initial"),
("passbook_stages_user_write", "0001_initial"),
("passbook_stages_user_login", "0001_initial"),
]
operations = [
migrations.RunPython(create_default_source_enrollment_flow),
migrations.RunPython(create_default_source_authentication_flow),
]

44
passbook/flows/migrations/0005_provider_flows.py Normal file
View file

@ -0,0 +1,44 @@
# Generated by Django 3.0.6 on 2020-05-24 11:34
from django.apps.registry import Apps
from django.db import migrations
from django.db.backends.base.schema import BaseDatabaseSchemaEditor
from passbook.flows.models import FlowDesignation
def create_default_provider_authz_flow(
apps: Apps, schema_editor: BaseDatabaseSchemaEditor
):
Flow = apps.get_model("passbook_flows", "Flow")
FlowStageBinding = apps.get_model("passbook_flows", "FlowStageBinding")
ConsentStage = apps.get_model("passbook_stages_consent", "ConsentStage")
db_alias = schema_editor.connection.alias
# Empty flow for providers where no consent is needed
Flow.objects.create(
name="default-provider-authorization",
slug="default-provider-authorization",
designation=FlowDesignation.AUTHORIZATION,
)
# Flow with consent form to obtain user consent for authorization
flow = Flow.objects.create(
name="default-provider-authorization-consent",
slug="default-provider-authorization-consent",
designation=FlowDesignation.AUTHORIZATION,
)
stage = ConsentStage.objects.create(name="default-provider-authorization-consent")
FlowStageBinding.objects.create(flow=flow, stage=stage, order=0)
class Migration(migrations.Migration):
dependencies = [
("passbook_flows", "0004_source_flows"),
("passbook_stages_consent", "0001_initial"),
]
operations = [migrations.RunPython(create_default_provider_authz_flow)]

12
passbook/flows/models.py
View file

@ -1,5 +1,5 @@
"""Flow models"""
from typing import Optional
from typing import Callable, Optional
from uuid import uuid4
from django.db import models
@ -9,6 +9,7 @@ from model_utils.managers import InheritanceManager
from structlog import get_logger
from passbook.core.types import UIUserSettings
from passbook.lib.utils.reflection import class_to_path
from passbook.policies.models import PolicyBindingModel
LOGGER = get_logger()
@ -19,6 +20,7 @@ class FlowDesignation(models.TextChoices):
should be replaced by a database entry."""
AUTHENTICATION = "authentication"
AUTHORIZATION = "authorization"
INVALIDATION = "invalidation"
ENROLLMENT = "enrollment"
UNRENOLLMENT = "unenrollment"
@ -48,6 +50,14 @@ class Stage(models.Model):
return f"Stage {self.name}"
def in_memory_stage(_type: Callable) -> Stage:
"""Creates an in-memory stage instance, based on a `_type` as view."""
class_path = class_to_path(_type)
stage = Stage()
stage.type = class_path
return stage
class Flow(PolicyBindingModel):
"""Flow describes how a series of Stages should be executed to authenticate/enroll/recover
a user. Additionally, policies can be applied, to specify which users

8
passbook/flows/planner.py
View file

@ -16,6 +16,7 @@ LOGGER = get_logger()
PLAN_CONTEXT_PENDING_USER = "pending_user"
PLAN_CONTEXT_SSO = "is_sso"
PLAN_CONTEXT_APPLICATION = "application"
def cache_key(flow: Flow, user: Optional[User] = None) -> str:
@ -45,10 +46,13 @@ class FlowPlanner:
that should be applied."""
use_cache: bool
allow_empty_flows: bool
flow: Flow
def __init__(self, flow: Flow):
self.use_cache = True
self.allow_empty_flows = False
self.flow = flow
def plan(
@ -80,11 +84,13 @@ class FlowPlanner:
LOGGER.debug(
"f(plan): Taking plan from cache", flow=self.flow, key=cached_plan_key
)
# Reset the context as this isn't factored into caching
cached_plan.context = default_context or {}
return cached_plan
LOGGER.debug("f(plan): building plan", flow=self.flow)
plan = self._build_plan(user, request, default_context)
cache.set(cache_key(self.flow, user), plan)
if not plan.stages:
if not plan.stages and not self.allow_empty_flows:
raise EmptyFlowException()
return plan

51
passbook/flows/templates/flows/shell.html
View file

@ -113,19 +113,18 @@ const updateMessages = () => {
});
});
};
const updateCard = (response) => {
if (!response.ok) {
console.log("well");
}
if (response.redirected && !response.url.endsWith(flowBodyUrl)) {
window.location = response.url;
} else {
response.text().then(text => {
flowBody.innerHTML = text;
const updateCard = (data) => {
switch (data.type) {
case "redirect":
window.location = data.to
break;
case "template":
flowBody.innerHTML = data.body;
updateMessages();
loadFormCode();
setFormSubmitHandlers();
});
default:
break;
}
};
const showSpinner = () => {
@ -139,10 +138,28 @@ const loadFormCode = () => {
document.head.appendChild(newScript);
});
};
const updateFormAction = (form) => {
for (let index = 0; index < form.elements.length; index++) {
const element = form.elements[index];
if (element.value === form.action) {
console.log("Found Form action URL in form elements, not changing form action.");
return false;
}
}
form.action = flowBodyUrl;
return true;
};
const checkAutosubmit = (form) => {
if ("autosubmit" in form.attributes) {
return form.submit();
}
};
const setFormSubmitHandlers = () => {
document.querySelectorAll("#flow-body form").forEach(form => {
console.log(`Checking for autosubmit attribute ${form}`);
checkAutosubmit(form);
console.log(`Setting action for form ${form}`);
form.action = flowBodyUrl;
updateFormAction(form);
console.log(`Adding handler for form ${form}`);
form.addEventListener('submit', (e) => {
e.preventDefault();
@ -150,19 +167,13 @@ const setFormSubmitHandlers = () => {
fetch(flowBodyUrl, {
method: 'post',
body: formData,
}).then((response) => {
showSpinner();
if (!response.url.endsWith(flowBodyUrl)) {
window.location = response.url;
} else {
updateCard(response);
}
}).then(response => response.json()).then(data => {
updateCard(data);
});
});
});
};
fetch(flowBodyUrl).then(updateCard);
fetch(flowBodyUrl).then(response => response.json()).then(data => updateCard(data));
</script>
{% endblock %}

36
passbook/flows/views.py
View file

@ -1,8 +1,15 @@
"""passbook multi-stage authentication engine"""
from typing import Any, Dict, Optional
from django.http import Http404, HttpRequest, HttpResponse
from django.http import (
Http404,
HttpRequest,
HttpResponse,
HttpResponseRedirect,
JsonResponse,
)
from django.shortcuts import get_object_or_404, redirect, reverse
from django.template.response import TemplateResponse
from django.utils.decorators import method_decorator
from django.views.decorators.clickjacking import xframe_options_sameorigin
from django.views.generic import TemplateView, View
@ -81,6 +88,8 @@ class FlowExecutorView(View):
)
stage_cls = path_to_class(self.current_stage.type)
self.current_stage_view = stage_cls(self)
self.current_stage_view.args = self.args
self.current_stage_view.kwargs = self.kwargs
self.current_stage_view.request = request
return super().dispatch(request)
@ -91,7 +100,8 @@ class FlowExecutorView(View):
view_class=class_to_path(self.current_stage_view.__class__),
flow_slug=self.flow.slug,
)
return self.current_stage_view.get(request, *args, **kwargs)
stage_response = self.current_stage_view.get(request, *args, **kwargs)
return to_stage_response(request, stage_response)
def post(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""pass post request to current stage"""
@ -100,7 +110,8 @@ class FlowExecutorView(View):
view_class=class_to_path(self.current_stage_view.__class__),
flow_slug=self.flow.slug,
)
return self.current_stage_view.post(request, *args, **kwargs)
stage_response = self.current_stage_view.post(request, *args, **kwargs)
return to_stage_response(request, stage_response)
def _initiate_plan(self) -> FlowPlan:
planner = FlowPlanner(self.flow)
@ -191,3 +202,22 @@ class FlowExecutorShellView(TemplateView):
kwargs["exec_url"] = reverse("passbook_flows:flow-executor", kwargs=self.kwargs)
kwargs["msg_url"] = reverse("passbook_api:messages-list")
return kwargs
def to_stage_response(request: HttpRequest, source: HttpResponse) -> HttpResponse:
"""Convert normal HttpResponse into JSON Response"""
if isinstance(source, HttpResponseRedirect) or source.status_code == 302:
redirect_url = source["Location"]
if request.path != redirect_url:
return JsonResponse({"type": "redirect", "to": redirect_url})
return source
if isinstance(source, TemplateResponse):
return JsonResponse(
{"type": "template", "body": source.render().content.decode("utf-8")}
)
# Check for actual HttpResponse (without isinstance as we dont want to check inheritance)
if source.__class__ == HttpResponse:
return JsonResponse(
{"type": "template", "body": source.content.decode("utf-8")}
)
return source

3
passbook/policies/types.py
View file

@ -38,6 +38,9 @@ class PolicyResult:
self.passing = passing
self.messages = messages
def __repr__(self):
return self.__str__()
def __str__(self):
if self.messages:
return f"PolicyResult passing={self.passing} messages={self.messages}"

2
passbook/providers/app_gw/forms.py
View file

@ -32,7 +32,7 @@ class ApplicationGatewayProviderForm(forms.ModelForm):
class Meta:
model = ApplicationGatewayProvider
fields = ["name", "internal_host", "external_host"]
fields = ["name", "authorization_flow", "internal_host", "external_host"]
widgets = {
"name": forms.TextInput(),
"internal_host": forms.TextInput(),

3
passbook/providers/app_gw/models.py
View file

@ -14,7 +14,8 @@ from passbook.lib.utils.template import render_to_string
class ApplicationGatewayProvider(Provider):
"""This provider uses oauth2_proxy with the OIDC Provider."""
"""Protect applications that don't support any of the other
Protocols by using a Reverse-Proxy."""
name = models.TextField()
internal_host = models.TextField()

11
passbook/providers/oauth/forms.py
View file

@ -1,21 +1,32 @@
"""passbook OAuth2 Provider Forms"""
from django import forms
from django.utils.translation import gettext_lazy as _
from passbook.flows.models import Flow, FlowDesignation
from passbook.providers.oauth.models import OAuth2Provider
class OAuth2ProviderForm(forms.ModelForm):
"""OAuth2 Provider form"""
authorization_flow = forms.ModelChoiceField(
queryset=Flow.objects.filter(designation=FlowDesignation.AUTHORIZATION)
)
class Meta:
model = OAuth2Provider
fields = [
"name",
"authorization_flow",
"redirect_uris",
"client_type",
"authorization_grant_type",
"client_id",
"client_secret",
]
labels = {
"client_id": _("Client ID"),
"redirect_uris": _("Redirect URIs"),
}

3
passbook/providers/oauth/models.py
View file

@ -12,7 +12,8 @@ from passbook.lib.utils.template import render_to_string
class OAuth2Provider(Provider, AbstractApplication):
"""Associate an OAuth2 Application with a Product"""
"""Generic OAuth2 Provider for applications not using OpenID-Connect. This Provider
also supports the GitHub-pretend mode."""
form = "passbook.providers.oauth.forms.OAuth2ProviderForm"

1
passbook/providers/oauth/settings.py
View file

@ -24,6 +24,7 @@ OAUTH2_PROVIDER = {
"SCOPES": {
"openid": "Access OpenID Userinfo",
"openid:userinfo": "Access OpenID Userinfo",
"email": "Access OpenID E-Mail",
# 'write': 'Write scope',
# 'groups': 'Access to your groups',
"user:email": "GitHub Compatibility: User E-Mail",

10
passbook/providers/oauth/urls.py
View file

@ -9,14 +9,9 @@ oauth_urlpatterns = [
# Custom OAuth2 Authorize View
path(
"authorize/",
oauth2.PassbookAuthorizationView.as_view(),
oauth2.AuthorizationFlowInitView.as_view(),
name="oauth2-authorize",
),
path(
"authorize/permission_denied/",
oauth2.OAuthPermissionDenied.as_view(),
name="oauth2-permission-denied",
),
# OAuth API
path("token/", views.TokenView.as_view(), name="token"),
path("revoke_token/", views.RevokeTokenView.as_view(), name="revoke-token"),
@ -26,7 +21,7 @@ oauth_urlpatterns = [
github_urlpatterns = [
path(
"login/oauth/authorize",
oauth2.PassbookAuthorizationView.as_view(),
oauth2.AuthorizationFlowInitView.as_view(),
name="github-authorize",
),
path(
@ -35,6 +30,7 @@ github_urlpatterns = [
name="github-access-token",
),
path("user", github.GitHubUserView.as_view(), name="github-user"),
path("user/teams", github.GitHubUserTeamsView.as_view(), name="github-user-teams"),
]
urlpatterns = [

39
passbook/providers/oauth/views/github.py
View file

@ -1,21 +1,32 @@
"""passbook pretend GitHub Views"""
from django.http import JsonResponse
from django.core.exceptions import PermissionDenied
from django.http import HttpRequest, HttpResponse, JsonResponse
from django.shortcuts import get_object_or_404
from django.views import View
from oauth2_provider.models import AccessToken
from passbook.core.models import User
class GitHubUserView(View):
class GitHubPretendView(View):
"""Emulate GitHub's API Endpoints"""
def verify_access_token(self) -> User:
"""Verify access token manually since github uses /user?access_token=..."""
if "HTTP_AUTHORIZATION" in self.request.META:
full_token = self.request.META.get("HTTP_AUTHORIZATION")
_, token = full_token.split(" ")
elif "access_token" in self.request.GET:
token = self.request.GET.get("access_token", "")
else:
raise PermissionDenied("No access token passed.")
return get_object_or_404(AccessToken, token=token).user
class GitHubUserView(GitHubPretendView):
"""Emulate GitHub's /user API Endpoint"""
def verify_access_token(self):
"""Verify access token manually since github uses /user?access_token=..."""
token = get_object_or_404(
AccessToken, token=self.request.GET.get("access_token", "")
)
return token.user
def get(self, request):
def get(self, request: HttpRequest) -> HttpResponse:
"""Emulate GitHub's /user API Endpoint"""
user = self.verify_access_token()
return JsonResponse(
@ -65,3 +76,11 @@ class GitHubUserView(View):
},
}
)
class GitHubUserTeamsView(GitHubPretendView):
"""Emulate GitHub's /user/teams API Endpoint"""
def get(self, request: HttpRequest) -> HttpResponse:
"""Emulate GitHub's /user/teams API Endpoint"""
return JsonResponse([], safe=False)

144
passbook/providers/oauth/views/oauth2.py
View file

@ -1,76 +1,124 @@
"""passbook OAuth2 Views"""
from typing import Optional
from urllib.parse import urlencode
from django.contrib import messages
from django.forms import Form
from django.http import HttpRequest, HttpResponse
from django.shortcuts import get_object_or_404, redirect, reverse
from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
from django.shortcuts import get_object_or_404, redirect
from django.views import View
from oauth2_provider.exceptions import OAuthToolkitError
from oauth2_provider.views.base import AuthorizationView
from structlog import get_logger
from passbook.audit.models import Event, EventAction
from passbook.core.models import Application
from passbook.core.views.access import AccessMixin
from passbook.core.views.utils import PermissionDeniedView
from passbook.flows.models import in_memory_stage
from passbook.flows.planner import (
PLAN_CONTEXT_APPLICATION,
PLAN_CONTEXT_SSO,
FlowPlanner,
)
from passbook.flows.stage import StageView
from passbook.flows.views import SESSION_KEY_PLAN
from passbook.lib.utils.urls import redirect_with_qs
from passbook.providers.oauth.models import OAuth2Provider
LOGGER = get_logger()
PLAN_CONTEXT_CLIENT_ID = "client_id"
PLAN_CONTEXT_REDIRECT_URI = "redirect_uri"
PLAN_CONTEXT_RESPONSE_TYPE = "response_type"
PLAN_CONTEXT_STATE = "state"
class OAuthPermissionDenied(PermissionDeniedView):
"""Show permission denied view"""
PLAN_CONTEXT_CODE_CHALLENGE = "code_challenge"
PLAN_CONTEXT_CODE_CHALLENGE_METHOD = "code_challenge_method"
PLAN_CONTEXT_SCOPE = "scope"
PLAN_CONTEXT_NONCE = "nonce"
class PassbookAuthorizationView(AccessMixin, AuthorizationView):
"""Custom OAuth2 Authorization View which checks policies, etc"""
class AuthorizationFlowInitView(AccessMixin, View):
"""OAuth2 Flow initializer, checks access to application and starts flow"""
_application: Optional[Application] = None
def _inject_response_type(self):
"""Inject response_type into querystring if not set"""
LOGGER.debug("response_type not set, defaulting to 'code'")
querystring = urlencode(self.request.GET)
querystring += "&response_type=code"
return redirect(
reverse("passbook_providers_oauth:oauth2-ok-authorize") + "?" + querystring
)
def dispatch(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""Update OAuth2Provider's skip_authorization state"""
# Get client_id to get provider, so we can update skip_authorization field
# pylint: disable=unused-argument
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""Check access to application, start FlowPLanner, return to flow executor shell"""
client_id = request.GET.get("client_id")
provider = get_object_or_404(OAuth2Provider, client_id=client_id)
try:
application = self.provider_to_application(provider)
except Application.DoesNotExist:
return redirect("passbook_providers_oauth:oauth2-permission-denied")
# Update field here so oauth-toolkit does work for us
provider.skip_authorization = application.skip_authorization
provider.save()
self._application = application
# Check permissions
result = self.user_has_access(self._application, request.user)
result = self.user_has_access(application, request.user)
if not result.passing:
for policy_message in result.messages:
messages.error(request, policy_message)
return redirect("passbook_providers_oauth:oauth2-permission-denied")
# Some clients don't pass response_type, so we default to code
if "response_type" not in request.GET:
return self._inject_response_type()
actual_response = AuthorizationView.dispatch(self, request, *args, **kwargs)
if actual_response.status_code == 400:
LOGGER.debug("Bad request", redirect_uri=request.GET.get("redirect_uri"))
return actual_response
def form_valid(self, form: Form):
# User has clicked on "Authorize"
Event.new(
EventAction.AUTHORIZE_APPLICATION, authorized_application=self._application,
).from_http(self.request)
LOGGER.debug(
"User authorized Application",
user=self.request.user,
application=self._application,
# Regardless, we start the planner and return to it
planner = FlowPlanner(provider.authorization_flow)
# planner.use_cache = False
planner.allow_empty_flows = True
plan = planner.plan(
self.request,
{
PLAN_CONTEXT_SSO: True,
PLAN_CONTEXT_APPLICATION: application,
PLAN_CONTEXT_CLIENT_ID: client_id,
PLAN_CONTEXT_REDIRECT_URI: request.GET.get(PLAN_CONTEXT_REDIRECT_URI),
PLAN_CONTEXT_RESPONSE_TYPE: request.GET.get(PLAN_CONTEXT_RESPONSE_TYPE),
PLAN_CONTEXT_STATE: request.GET.get(PLAN_CONTEXT_STATE),
PLAN_CONTEXT_SCOPE: request.GET.get(PLAN_CONTEXT_SCOPE),
PLAN_CONTEXT_NONCE: request.GET.get(PLAN_CONTEXT_NONCE),
},
)
return AuthorizationView.form_valid(self, form)
plan.stages.append(in_memory_stage(OAuth2Stage))
self.request.session[SESSION_KEY_PLAN] = plan
return redirect_with_qs(
"passbook_flows:flow-executor-shell",
self.request.GET,
flow_slug=provider.authorization_flow.slug,
)
class OAuth2Stage(AuthorizationView, StageView):
"""OAuth2 Stage, dynamically injected into the plan"""
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""Last stage in flow, finalizes OAuth Response and redirects to Client"""
application: Application = self.executor.plan.context[PLAN_CONTEXT_APPLICATION]
provider: OAuth2Provider = application.provider
Event.new(
EventAction.AUTHORIZE_APPLICATION, authorized_application=application,
).from_http(self.request)
credentials = {
"client_id": self.executor.plan.context[PLAN_CONTEXT_CLIENT_ID],
"redirect_uri": self.executor.plan.context[PLAN_CONTEXT_REDIRECT_URI],
"response_type": self.executor.plan.context.get(
PLAN_CONTEXT_RESPONSE_TYPE, None
),
"state": self.executor.plan.context.get(PLAN_CONTEXT_STATE, None),
"nonce": self.executor.plan.context.get(PLAN_CONTEXT_NONCE, None),
}
if self.executor.plan.context.get(PLAN_CONTEXT_CODE_CHALLENGE, False):
credentials[PLAN_CONTEXT_CODE_CHALLENGE] = self.executor.plan.context.get(
PLAN_CONTEXT_CODE_CHALLENGE
)
if self.executor.plan.context.get(PLAN_CONTEXT_CODE_CHALLENGE_METHOD, False):
credentials[
PLAN_CONTEXT_CODE_CHALLENGE_METHOD
] = self.executor.plan.context.get(PLAN_CONTEXT_CODE_CHALLENGE_METHOD)
scopes = self.executor.plan.context.get(PLAN_CONTEXT_SCOPE)
try:
uri, _headers, _body, _status = self.create_authorization_response(
request=self.request,
scopes=scopes,
credentials=credentials,
allow=True,
)
LOGGER.debug("Success url for the request: {0}".format(uri))
except OAuthToolkitError as error:
return self.error_response(error, provider)
self.executor.stage_ok()
return HttpResponseRedirect(self.redirect(uri, provider).url)

5
passbook/providers/oidc/apps.py
View file

@ -1,6 +1,4 @@
"""passbook auth oidc provider app config"""
from importlib import import_module
from django.apps import AppConfig
from django.db.utils import InternalError, OperationalError, ProgrammingError
from django.urls import include, path
@ -15,6 +13,7 @@ class PassbookProviderOIDCConfig(AppConfig):
name = "passbook.providers.oidc"
label = "passbook_providers_oidc"
verbose_name = "passbook Providers.OIDC"
mountpoint = "application/oidc/"
def ready(self):
try:
@ -36,5 +35,3 @@ class PassbookProviderOIDCConfig(AppConfig):
include("oidc_provider.urls", namespace="oidc_provider"),
),
)
import_module("passbook.providers.oidc.signals")

7
passbook/providers/oidc/auth.py
View file

@ -10,6 +10,8 @@ from structlog import get_logger
from passbook.audit.models import Event, EventAction
from passbook.core.models import Application, Provider, User
from passbook.flows.planner import FlowPlan
from passbook.flows.views import SESSION_KEY_PLAN
from passbook.policies.engine import PolicyEngine
LOGGER = get_logger()
@ -46,7 +48,7 @@ def check_permissions(
LOGGER.debug(
"Checking permissions for application", user=user, application=application
)
policy_engine = PolicyEngine(application.policies.all(), user, request)
policy_engine = PolicyEngine(application, user, request)
policy_engine.build()
# Check permissions
@ -56,9 +58,10 @@ def check_permissions(
messages.error(request, policy_message)
return redirect("passbook_providers_oauth:oauth2-permission-denied")
plan: FlowPlan = request.session[SESSION_KEY_PLAN]
Event.new(
EventAction.AUTHORIZE_APPLICATION,
authorized_application=application,
skipped_authorization=False,
flow=plan.flow_pk,
).from_http(request)
return None

24
passbook/providers/oidc/forms.py
View file

@ -4,12 +4,17 @@ from django import forms
from oauth2_provider.generators import generate_client_id, generate_client_secret
from oidc_provider.models import Client
from passbook.flows.models import Flow, FlowDesignation
from passbook.providers.oidc.models import OpenIDProvider
class OIDCProviderForm(forms.ModelForm):
"""OpenID Client form"""
authorization_flow = forms.ModelChoiceField(
queryset=Flow.objects.filter(designation=FlowDesignation.AUTHORIZATION)
)
def __init__(self, *args, **kwargs):
# Correctly load data from 1:1 rel
if "instance" in kwargs and kwargs["instance"]:
@ -17,20 +22,35 @@ class OIDCProviderForm(forms.ModelForm):
super().__init__(*args, **kwargs)
self.fields["client_id"].initial = generate_client_id()
self.fields["client_secret"].initial = generate_client_secret()
try:
self.fields[
"authorization_flow"
].initial = self.instance.openidprovider.authorization_flow
# pylint: disable=no-member
except Client.openidprovider.RelatedObjectDoesNotExist:
pass
def save(self, *args, **kwargs):
self.instance.reuse_consent = False # This is managed by passbook
self.instance.require_consent = True # This is managed by passbook
self.instance.require_consent = False # This is managed by passbook
response = super().save(*args, **kwargs)
# Check if openidprovider class instance exists
if not OpenIDProvider.objects.filter(oidc_client=self.instance).exists():
OpenIDProvider.objects.create(oidc_client=self.instance)
OpenIDProvider.objects.create(
oidc_client=self.instance,
authorization_flow=self.cleaned_data.get("authorization_flow"),
)
self.instance.openidprovider.authorization_flow = self.cleaned_data.get(
"authorization_flow"
)
self.instance.openidprovider.save()
return response
class Meta:
model = Client
fields = [
"name",
"authorization_flow",
"client_type",
"client_id",
"client_secret",

8
passbook/providers/oidc/models.py
View file

@ -12,7 +12,7 @@ from passbook.lib.utils.template import render_to_string
class OpenIDProvider(Provider):
"""Proxy model for OIDC Client"""
"""OpenID Connect Provider for applications that support OIDC."""
# Since oidc_provider doesn't currently support swappable models
# (https://github.com/juanifioren/django-oidc-provider/pull/305)
@ -28,7 +28,7 @@ class OpenIDProvider(Provider):
return self.oidc_client.name
def __str__(self):
return "OpenID Connect Provider %s" % self.oidc_client.__str__()
return f"OpenID Connect Provider {self.oidc_client.__str__()}"
def html_setup_urls(self, request: HttpRequest) -> Optional[str]:
"""return template and context modal with URLs for authorize, token, openid-config, etc"""
@ -37,14 +37,14 @@ class OpenIDProvider(Provider):
{
"provider": self,
"authorize": request.build_absolute_uri(
reverse("oidc_provider:authorize")
reverse("passbook_providers_oidc:authorize")
),
"token": request.build_absolute_uri(reverse("oidc_provider:token")),
"userinfo": request.build_absolute_uri(
reverse("oidc_provider:userinfo")
),
"provider_info": request.build_absolute_uri(
reverse("oidc_provider:provider-info")
reverse("passbook_providers_oidc:provider-info")
),
},
)

15
passbook/providers/oidc/signals.py
View file

@ -1,15 +0,0 @@
"""OIDC Provider signals"""
from django.db.models.signals import post_save
from django.dispatch import receiver
from passbook.core.models import Application
from passbook.providers.oidc.models import OpenIDProvider
@receiver(post_save, sender=Application)
# pylint: disable=unused-argument
def on_application_save(sender, instance: Application, **_):
"""Synchronize application's skip_authorization with oidc_client's require_consent"""
if isinstance(instance.provider, OpenIDProvider):
instance.provider.oidc_client.require_consent = not instance.skip_authorization
instance.provider.oidc_client.save()

13
passbook/providers/oidc/urls.py Normal file
View file

@ -0,0 +1,13 @@
"""oidc provider URLs"""
from django.conf.urls import url
from passbook.providers.oidc.views import AuthorizationFlowInitView, ProviderInfoView
urlpatterns = [
url(r"^authorize/?$", AuthorizationFlowInitView.as_view(), name="authorize"),
url(
r"^\.well-known/openid-configuration/?$",
ProviderInfoView.as_view(),
name="provider-info",
),
]

127
passbook/providers/oidc/views.py Normal file
View file

@ -0,0 +1,127 @@
"""passbook OIDC Views"""
from django.contrib import messages
from django.http import HttpRequest, HttpResponse, JsonResponse
from django.shortcuts import get_object_or_404, redirect, reverse
from django.views import View
from oidc_provider.lib.endpoints.authorize import AuthorizeEndpoint
from oidc_provider.lib.utils.common import get_issuer, get_site_url
from oidc_provider.models import ResponseType
from oidc_provider.views import AuthorizeView
from structlog import get_logger
from passbook.core.models import Application
from passbook.core.views.access import AccessMixin
from passbook.flows.models import in_memory_stage
from passbook.flows.planner import (
PLAN_CONTEXT_APPLICATION,
PLAN_CONTEXT_SSO,
FlowPlan,
FlowPlanner,
)
from passbook.flows.stage import StageView
from passbook.flows.views import SESSION_KEY_PLAN
from passbook.lib.utils.urls import redirect_with_qs
from passbook.providers.oidc.models import OpenIDProvider
LOGGER = get_logger()
PLAN_CONTEXT_PARAMS = "params"
class AuthorizationFlowInitView(AccessMixin, View):
"""OIDC Flow initializer, checks access to application and starts flow"""
# pylint: disable=unused-argument
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""Check access to application, start FlowPLanner, return to flow executor shell"""
client_id = request.GET.get("client_id")
provider = get_object_or_404(OpenIDProvider, oidc_client__client_id=client_id)
try:
application = self.provider_to_application(provider)
except Application.DoesNotExist:
return redirect("passbook_providers_oauth:oauth2-permission-denied")
# Check permissions
result = self.user_has_access(application, request.user)
if not result.passing:
for policy_message in result.messages:
messages.error(request, policy_message)
return redirect("passbook_providers_oauth:oauth2-permission-denied")
# Extract params so we can save them in the plan context
endpoint = AuthorizeEndpoint(request)
# Regardless, we start the planner and return to it
planner = FlowPlanner(provider.authorization_flow)
# planner.use_cache = False
planner.allow_empty_flows = True
plan = planner.plan(
self.request,
{
PLAN_CONTEXT_SSO: True,
PLAN_CONTEXT_APPLICATION: application,
PLAN_CONTEXT_PARAMS: endpoint.params,
},
)
plan.stages.append(in_memory_stage(OIDCStage))
self.request.session[SESSION_KEY_PLAN] = plan
return redirect_with_qs(
"passbook_flows:flow-executor-shell",
self.request.GET,
flow_slug=provider.authorization_flow.slug,
)
class FlowAuthorizeEndpoint(AuthorizeEndpoint):
"""Restore params from flow context"""
def _extract_params(self):
plan: FlowPlan = self.request.session[SESSION_KEY_PLAN]
self.params = plan.context[PLAN_CONTEXT_PARAMS]
class OIDCStage(AuthorizeView, StageView):
"""Finall stage, restores params from Flow."""
authorize_endpoint_class = FlowAuthorizeEndpoint
class ProviderInfoView(View):
"""Custom ProviderInfo View which shows our URLs instead"""
# pylint: disable=unused-argument
def get(self, request, *args, **kwargs):
"""Custom ProviderInfo View which shows our URLs instead"""
dic = dict()
site_url = get_site_url(request=request)
dic["issuer"] = get_issuer(site_url=site_url, request=request)
dic["authorization_endpoint"] = site_url + reverse(
"passbook_providers_oidc:authorize"
)
dic["token_endpoint"] = site_url + reverse("oidc_provider:token")
dic["userinfo_endpoint"] = site_url + reverse("oidc_provider:userinfo")
dic["end_session_endpoint"] = site_url + reverse("oidc_provider:end-session")
dic["introspection_endpoint"] = site_url + reverse(
"oidc_provider:token-introspection"
)
types_supported = [
response_type.value for response_type in ResponseType.objects.all()
]
dic["response_types_supported"] = types_supported
dic["jwks_uri"] = site_url + reverse("oidc_provider:jwks")
dic["id_token_signing_alg_values_supported"] = ["HS256", "RS256"]
# See: http://openid.net/specs/openid-connect-core-1_0.html#SubjectIDTypes
dic["subject_types_supported"] = ["public"]
dic["token_endpoint_auth_methods_supported"] = [
"client_secret_post",
"client_secret_basic",
]
response = JsonResponse(dic)
response["Access-Control-Allow-Origin"] = "*"
return response

6
passbook/providers/saml/forms.py
View file

@ -5,6 +5,7 @@ from django.contrib.admin.widgets import FilteredSelectMultiple
from django.utils.translation import gettext as _
from passbook.core.expression import PropertyMappingEvaluator
from passbook.flows.models import Flow, FlowDesignation
from passbook.providers.saml.models import (
SAMLPropertyMapping,
SAMLProvider,
@ -15,6 +16,9 @@ from passbook.providers.saml.models import (
class SAMLProviderForm(forms.ModelForm):
"""SAML Provider form"""
authorization_flow = forms.ModelChoiceField(
queryset=Flow.objects.filter(designation=FlowDesignation.AUTHORIZATION)
)
processor_path = forms.ChoiceField(
choices=get_provider_choices(), label="Processor"
)
@ -24,10 +28,12 @@ class SAMLProviderForm(forms.ModelForm):
model = SAMLProvider
fields = [
"name",
"authorization_flow",
"processor_path",
"acs_url",
"audience",
"issuer",
"sp_binding",
"assertion_valid_not_before",
"assertion_valid_not_on_or_after",
"session_valid_not_on_or_after",

12
passbook/providers/saml/migrations/0002_default_saml_property_mappings.py
View file

@ -13,32 +13,32 @@ def create_default_property_mappings(apps, schema_editor):
{
"FriendlyName": "eduPersonPrincipalName",
"Name": "urn:oid:1.3.6.1.4.1.5923.1.1.1.6",
"Expression": "return user.get('email')",
"Expression": "return user.email",
},
{
"FriendlyName": "cn",
"Name": "urn:oid:2.5.4.3",
"Expression": "return user.get('name')",
"Expression": "return user.name",
},
{
"FriendlyName": "mail",
"Name": "urn:oid:0.9.2342.19200300.100.1.3",
"Expression": "return user.get('email')",
"Expression": "return user.email",
},
{
"FriendlyName": "displayName",
"Name": "urn:oid:2.16.840.1.113730.3.1.241",
"Expression": "return user.get('username')",
"Expression": "return user.username",
},
{
"FriendlyName": "uid",
"Name": "urn:oid:0.9.2342.19200300.100.1.1",
"Expression": "return user.get('pk')",
"Expression": "return user.pk",
},
{
"FriendlyName": "member-of",
"Name": "member-of",
"Expression": "[{% for group in user.groups.all() %}'{{ group.name }}',{% endfor %}]",
"Expression": "for group in user.groups.all():\n yield group.name",
},
]
for default in defaults:

20
passbook/providers/saml/migrations/0003_samlprovider_sp_binding.py Normal file
View file

@ -0,0 +1,20 @@
# Generated by Django 3.0.6 on 2020-06-06 13:25
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("passbook_providers_saml", "0002_default_saml_property_mappings"),
]
operations = [
migrations.AddField(
model_name="samlprovider",
name="sp_binding",
field=models.TextField(
choices=[("redirect", "Redirect"), ("post", "Post")], default="redirect"
),
),
]

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

@ -17,6 +17,13 @@ from passbook.providers.saml.utils.time import timedelta_string_validator
LOGGER = get_logger()
class SAMLBindings(models.TextChoices):
"""SAML Bindings supported by passbook"""
REDIRECT = "redirect"
POST = "post"
class SAMLProvider(Provider):
"""Model to save information about a Remote SAML Endpoint"""
@ -26,6 +33,9 @@ class SAMLProvider(Provider):
acs_url = models.URLField(verbose_name=_("ACS URL"))
audience = models.TextField(default="")
issuer = models.TextField(help_text=_("Also known as EntityID"))
sp_binding = models.TextField(
choices=SAMLBindings.choices, default=SAMLBindings.REDIRECT
)
assertion_valid_not_before = models.TextField(
default="minutes=-5",
@ -118,8 +128,8 @@ class SAMLProvider(Provider):
try:
# pylint: disable=no-member
return reverse(
"passbook_providers_saml:saml-metadata",
kwargs={"application": self.application.slug},
"passbook_providers_saml:metadata",
kwargs={"application_slug": self.application.slug},
)
except Provider.application.RelatedObjectDoesNotExist:
return None

30
passbook/providers/saml/templates/saml/idp/autosubmit_form.html
View file

@ -4,30 +4,26 @@
{% load i18n %}
{% block card_title %}
{% trans 'Redirecting...' %}
{% blocktrans with app=application.name %}
Redirecting to {{ app }}...
{% endblocktrans %}
{% endblock %}
{% block card %}
<form method="POST" action="{{ url }}">
<form method="POST" action="{{ url }}" autosubmit>
{% csrf_token %}
{% for key, value in attrs.items %}
<input type="hidden" name="{{ key }}" value="{{ value }}">
{% endfor %}
<div class="login-group">
<p>
{% blocktrans with user=user %}
You are logged in as {{ user }}.
{% endblocktrans %}
<a href="{% url 'passbook_flows:default-invalidation' %}">{% trans 'Not you?' %}</a>
</p>
<input class="btn btn-primary btn-block btn-lg" type="submit" value="{% trans 'Continue' %}" />
<div class="pf-c-form__group">
<span class="pf-c-spinner" role="progressbar" aria-valuetext="Loading...">
<span class="pf-c-spinner__clipper"></span>
<span class="pf-c-spinner__lead-ball"></span>
<span class="pf-c-spinner__tail-ball"></span>
</span>
</div>
<div class="pf-c-form__group pf-m-action">
<button class="pf-c-button pf-m-primary pf-m-block" type="submit">{% trans 'Continue' %}</button>
</div>
</form>
{% endblock %}
{% block scripts %}
{{ block.super }}
<script>
document.querySelector("form").submit();
</script>
{% endblock %}

26
passbook/providers/saml/templates/saml/idp/login.html
View file

@ -1,26 +0,0 @@
{% extends "login/base.html" %}
{% load passbook_utils %}
{% load i18n %}
{% block card %}
<form method="POST" class="pf-c-form">
{% csrf_token %}
<div class="pf-c-form__group">
<h3>
{% blocktrans with provider=provider.application.name %}
You're about to sign into {{ provider }}
{% endblocktrans %}
</h3>
<p>
{% blocktrans with user=user %}
You are logged in as {{ user }}.
{% endblocktrans %}
<a href="{% url 'passbook_flows:default-invalidation' %}">{% trans 'Not you?' %}</a>
</p>
</div>
<div class="pf-c-form__group pf-m-action">
<input class="pf-c-button pf-m-primary pf-m-block" type="submit" value="{% trans 'Continue' %}" />
</div>
</form>
{% endblock %}

5
passbook/providers/saml/templates/saml/xml/metadata.xml
View file

@ -11,8 +11,7 @@
</md:KeyDescriptor>
{% endif %}
<md:NameIDFormat>{{ subject_format }}</md:NameIDFormat>
<md:SingleLogoutService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect" Location="{{ slo_url }}"/>
<md:SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST" Location="{{ sso_post_url }}"/>
<md:SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect" Location="{{ sso_redirect_url }}"/>
<md:SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST" Location="{{ sso_binding_post }}"/>
<md:SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect" Location="{{ sso_binding_redirect }}"/>
</md:IDPSSODescriptor>
</md:EntityDescriptor>

31
passbook/providers/saml/urls.py
View file

@ -4,31 +4,26 @@ from django.urls import path
from passbook.providers.saml import views
urlpatterns = [
# This view is used to initiate a Login-flow from the IDP
# SSO Bindings
path(
"<slug:application>/login/initiate/",
views.InitiateLoginView.as_view(),
name="saml-login-initiate",
),
# This view is the endpoint a SP would redirect to, and saves data into the session
# this is required as the process view which it redirects to might have to login first.
path(
"<slug:application>/login/", views.LoginBeginView.as_view(), name="saml-login"
"<slug:application_slug>/sso/binding/redirect/",
views.SAMLSSOBindingRedirectView.as_view(),
name="sso-redirect",
),
path(
"<slug:application>/login/authorize/",
views.AuthorizeView.as_view(),
name="saml-login-authorize",
"<slug:application_slug>/sso/binding/post/",
views.SAMLSSOBindingPOSTView.as_view(),
name="sso-post",
),
path("<slug:application>/logout/", views.LogoutView.as_view(), name="saml-logout"),
# SSO IdP Initiated
path(
"<slug:application>/logout/slo/",
views.SLOLogout.as_view(),
name="saml-logout-slo",
"<slug:application_slug>/sso/binding/init/",
views.SAMLSSOBindingInitView.as_view(),
name="sso-init",
),
path(
"<slug:application>/metadata/",
"<slug:application_slug>/metadata/",
views.DescriptorDownloadView.as_view(),
name="saml-metadata",
name="metadata",
),
]

349
passbook/providers/saml/views.py
View file

@ -1,15 +1,14 @@
"""passbook SAML IDP Views"""
from typing import Optional
from django.contrib.auth import logout
from django.contrib.auth.mixins import AccessMixin
from django.core.exceptions import ValidationError
from django.contrib import messages
from django.contrib.auth.mixins import LoginRequiredMixin
from django.core.exceptions import PermissionDenied
from django.core.validators import URLValidator
from django.http import HttpRequest, HttpResponse
from django.shortcuts import get_object_or_404, redirect, render, reverse
from django.utils.datastructures import MultiValueDictKeyError
from django.utils.decorators import method_decorator
from django.utils.html import mark_safe
from django.utils.http import urlencode
from django.utils.translation import gettext as _
from django.views import View
from django.views.decorators.csrf import csrf_exempt
@ -18,11 +17,20 @@ from structlog import get_logger
from passbook.audit.models import Event, EventAction
from passbook.core.models import Application, Provider
from passbook.flows.models import in_memory_stage
from passbook.flows.planner import (
PLAN_CONTEXT_APPLICATION,
PLAN_CONTEXT_SSO,
FlowPlanner,
)
from passbook.flows.stage import StageView
from passbook.flows.views import SESSION_KEY_PLAN
from passbook.lib.utils.template import render_to_string
from passbook.lib.utils.urls import redirect_with_qs
from passbook.lib.views import bad_request_message
from passbook.policies.engine import PolicyEngine
from passbook.providers.saml.exceptions import CannotHandleAssertion
from passbook.providers.saml.models import SAMLProvider
from passbook.providers.saml.models import SAMLBindings, SAMLProvider
from passbook.providers.saml.processors.types import SAMLResponseParams
LOGGER = get_logger()
@ -33,69 +41,82 @@ SESSION_KEY_RELAY_STATE = "RelayState"
SESSION_KEY_PARAMS = "SAMLParams"
class AccessRequiredView(AccessMixin, View):
"""Mixin class for Views using a provider instance"""
class SAMLAccessMixin:
"""SAML base access mixin, checks access to an application based on its policies"""
_provider: Optional[SAMLProvider] = None
@property
def provider(self) -> SAMLProvider:
"""Get provider instance"""
if not self._provider:
application = get_object_or_404(
Application, slug=self.kwargs["application"]
)
provider: SAMLProvider = get_object_or_404(
SAMLProvider, pk=application.provider_id
)
self._provider = provider
return self._provider
return self._provider
request: HttpRequest
application: Application
provider: SAMLProvider
def _has_access(self) -> bool:
"""Check if user has access to application"""
policy_engine = PolicyEngine(
self.provider.application.policies.all(), self.request.user, self.request
)
"""Check if user has access to application, add an error if not"""
policy_engine = PolicyEngine(self.application, self.request.user, self.request)
policy_engine.build()
passing = policy_engine.passing
result = policy_engine.result
LOGGER.debug(
"saml_has_access",
"SAMLFlowInit _has_access",
user=self.request.user,
app=self.provider.application,
passing=passing,
app=self.application,
result=result,
)
return passing
if not result.passing:
for message in result.messages:
messages.error(self.request, _(message))
return result.passing
def dispatch(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
if not request.user.is_authenticated:
return self.handle_no_permission()
class SAMLSSOView(LoginRequiredMixin, SAMLAccessMixin, View):
""""SAML SSO Base View, which plans a flow and injects our final stage.
Calls get/post handler."""
def dispatch(
self, request: HttpRequest, *args, application_slug: str, **kwargs
) -> HttpResponse:
self.application = get_object_or_404(Application, slug=application_slug)
self.provider: SAMLProvider = get_object_or_404(
SAMLProvider, pk=self.application.provider_id
)
if not self._has_access():
return render(
request,
"login/denied.html",
{"title": _("You don't have access to this application")},
raise PermissionDenied()
# Call the method handler, which checks the SAML Request
method_response = super().dispatch(request, *args, application_slug, **kwargs)
if method_response:
return method_response
# Regardless, we start the planner and return to it
planner = FlowPlanner(self.provider.authorization_flow)
planner.allow_empty_flows = True
plan = planner.plan(
self.request,
{PLAN_CONTEXT_SSO: True, PLAN_CONTEXT_APPLICATION: self.application},
)
plan.stages.append(in_memory_stage(SAMLFlowFinalView))
self.request.session[SESSION_KEY_PLAN] = plan
return redirect_with_qs(
"passbook_flows:flow-executor-shell",
self.request.GET,
flow_slug=self.provider.authorization_flow.slug,
)
return super().dispatch(request, *args, **kwargs)
class LoginBeginView(AccessRequiredView):
"""Receives a SAML 2.0 AuthnRequest from a Service Provider and
stores it in the session prior to enforcing login."""
class SAMLSSOBindingRedirectView(SAMLSSOView):
"""SAML Handler for SSO/Redirect bindings, which are sent via GET"""
def handler(self, source, application: str) -> HttpResponse:
"""Handle SAML Request whether its a POST or a Redirect binding"""
# pylint: disable=unused-argument
def get(
self, request: HttpRequest, application_slug: str
) -> Optional[HttpResponse]:
"""Handle REDIRECT bindings"""
# Store these values now, because Django's login cycle won't preserve them.
try:
self.request.session[SESSION_KEY_SAML_REQUEST] = source[
SESSION_KEY_SAML_REQUEST
]
except (KeyError, MultiValueDictKeyError):
if SESSION_KEY_SAML_REQUEST not in request.GET:
LOGGER.info("handle_saml_request: SAML payload missing")
return bad_request_message(
self.request, "The SAML request payload is missing."
)
self.request.session[SESSION_KEY_RELAY_STATE] = source.get(
self.request.session[SESSION_KEY_SAML_REQUEST] = request.GET[
SESSION_KEY_SAML_REQUEST
]
self.request.session[SESSION_KEY_RELAY_STATE] = request.GET.get(
SESSION_KEY_RELAY_STATE, ""
)
@ -105,104 +126,89 @@ class LoginBeginView(AccessRequiredView):
self.request.session[SESSION_KEY_PARAMS] = params
except CannotHandleAssertion as exc:
LOGGER.info(exc)
did_you_mean_link = self.request.build_absolute_uri(
reverse(
"passbook_providers_saml:saml-login-initiate",
kwargs={"application": application},
)
)
did_you_mean_message = (
f" Did you mean to go <a href='{did_you_mean_link}'>here</a>?"
)
return bad_request_message(
self.request, mark_safe(str(exc) + did_you_mean_message)
)
return redirect(
reverse(
"passbook_providers_saml:saml-login-authorize",
kwargs={"application": application},
)
)
@method_decorator(csrf_exempt)
def dispatch(self, *args, **kwargs):
return super().dispatch(*args, **kwargs)
@method_decorator(csrf_exempt)
def get(self, request: HttpRequest, application: str) -> HttpResponse:
"""Handle REDIRECT bindings"""
return self.handler(request.GET, application)
@method_decorator(csrf_exempt)
def post(self, request: HttpRequest, application: str) -> HttpResponse:
"""Handle POST Bindings"""
return self.handler(request.POST, application)
return bad_request_message(self.request, str(exc))
return None
class InitiateLoginView(AccessRequiredView):
"""IdP-initiated Login"""
def get(self, request: HttpRequest, application: str) -> HttpResponse:
"""Initiates an IdP-initiated link to a simple SP resource/target URL."""
self.provider.processor.is_idp_initiated = True
self.provider.processor.init_deep_link(request)
params = self.provider.processor.generate_response()
request.session[SESSION_KEY_PARAMS] = params
return redirect(
reverse(
"passbook_providers_saml:saml-login-authorize",
kwargs={"application": application},
)
)
class AuthorizeView(AccessRequiredView):
"""Ask the user for authorization to continue to the SP.
Presents a SAML 2.0 Assertion for POSTing back to the Service Provider."""
def get(self, request: HttpRequest, application: str) -> HttpResponse:
"""Handle get request, i.e. render form"""
# User access gets checked in dispatch
# Otherwise we generate the IdP initiated session
try:
# application.skip_authorization is set so we directly redirect the user
if self.provider.application.skip_authorization:
LOGGER.debug("skipping authz", application=self.provider.application)
return self.post(request, application)
return render(
request,
"saml/idp/login.html",
{"provider": self.provider, "title": "Authorize Application"},
)
except KeyError:
return bad_request_message(request, "Missing SAML Payload")
@method_decorator(csrf_exempt, name="dispatch")
class SAMLSSOBindingPOSTView(SAMLSSOView):
"""SAML Handler for SSO/POST bindings"""
# pylint: disable=unused-argument
def post(self, request: HttpRequest, application: str) -> HttpResponse:
"""Handle post request, return back to ACS"""
# User access gets checked in dispatch
def post(
self, request: HttpRequest, application_slug: str
) -> Optional[HttpResponse]:
"""Handle POST bindings"""
# Store these values now, because Django's login cycle won't preserve them.
if SESSION_KEY_SAML_REQUEST not in request.POST:
LOGGER.info("handle_saml_request: SAML payload missing")
return bad_request_message(
self.request, "The SAML request payload is missing."
)
# we get here when skip_authorization is True, and after the user accepted
# the authorization form
self.request.session[SESSION_KEY_SAML_REQUEST] = request.POST[
SESSION_KEY_SAML_REQUEST
]
self.request.session[SESSION_KEY_RELAY_STATE] = request.POST.get(
SESSION_KEY_RELAY_STATE, ""
)
try:
self.provider.processor.can_handle(self.request)
params = self.provider.processor.generate_response()
self.request.session[SESSION_KEY_PARAMS] = params
except CannotHandleAssertion as exc:
LOGGER.info(exc)
return bad_request_message(self.request, str(exc))
return None
class SAMLSSOBindingInitView(SAMLSSOView):
"""SAML Handler for for IdP Initiated login flows"""
# pylint: disable=unused-argument
def get(
self, request: HttpRequest, application_slug: str
) -> Optional[HttpResponse]:
"""Create saml params from scratch"""
LOGGER.debug(
"handle_saml_no_request: No SAML Request, using IdP-initiated flow."
)
self.provider.processor.is_idp_initiated = True
self.provider.processor.init_deep_link(self.request)
params = self.provider.processor.generate_response()
self.request.session[SESSION_KEY_PARAMS] = params
# This View doesn't have a URL on purpose, as its called by the FlowExecutor
class SAMLFlowFinalView(StageView):
"""View used by FlowExecutor after all stages have passed. Logs the authorization,
and redirects to the SP (if REDIRECT is configured) or shows and auto-submit for
(if POST is configured)."""
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
application: Application = self.executor.plan.context[PLAN_CONTEXT_APPLICATION]
provider: SAMLProvider = application.provider
# Log Application Authorization
Event.new(
EventAction.AUTHORIZE_APPLICATION,
authorized_application=self.provider.application,
skipped_authorization=self.provider.application.skip_authorization,
authorized_application=application,
flow=self.executor.plan.flow_pk,
).from_http(self.request)
self.request.session.pop(SESSION_KEY_SAML_REQUEST, None)
self.request.session.pop(SESSION_KEY_SAML_RESPONSE, None)
self.request.session.pop(SESSION_KEY_RELAY_STATE, None)
if SESSION_KEY_PARAMS not in self.request.session:
return self.executor.stage_invalid()
response: SAMLResponseParams = self.request.session.pop(SESSION_KEY_PARAMS)
if provider.sp_binding == SAMLBindings.POST:
return render(
self.request,
"saml/idp/autosubmit_form.html",
{
"url": response.acs_url,
"application": application,
"attrs": {
"ACSUrl": response.acs_url,
SESSION_KEY_SAML_RESPONSE: response.saml_response,
@ -210,77 +216,41 @@ class AuthorizeView(AccessRequiredView):
},
},
)
if provider.sp_binding == SAMLBindings.REDIRECT:
querystring = urlencode(
{
SESSION_KEY_SAML_RESPONSE: response.saml_response,
SESSION_KEY_RELAY_STATE: response.relay_state,
}
)
return redirect(f"{response.acs_url}?{querystring}")
return bad_request_message(request, "Invalid sp_binding specified")
@method_decorator(csrf_exempt, name="dispatch")
class LogoutView(AccessRequiredView):
"""Allows a non-SAML 2.0 URL to log out the user and
returns a standard logged-out page. (SalesForce and others use this method,
though it's technically not SAML 2.0)."""
# pylint: disable=unused-argument
def get(self, request: HttpRequest, application: str) -> HttpResponse:
"""Perform logout"""
logout(request)
redirect_url = request.GET.get("redirect_to", "")
try:
URL_VALIDATOR(redirect_url)
except ValidationError:
pass
else:
return redirect(redirect_url)
return render(request, "saml/idp/logged_out.html")
@method_decorator(csrf_exempt, name="dispatch")
class SLOLogout(AccessRequiredView):
"""Receives a SAML 2.0 LogoutRequest from a Service Provider,
logs out the user and returns a standard logged-out page."""
# pylint: disable=unused-argument
def post(self, request: HttpRequest, application: str) -> HttpResponse:
"""Perform logout"""
request.session[SESSION_KEY_SAML_REQUEST] = request.POST[
SESSION_KEY_SAML_REQUEST
]
# TODO: Parse SAML LogoutRequest from POST data, similar to login_process().
# TODO: Modify the base processor to handle logouts?
# TODO: Combine this with login_process(), since they are so very similar?
# TODO: Format a LogoutResponse and return it to the browser.
# XXX: For now, simply log out without validating the request.
logout(request)
return render(request, "saml/idp/logged_out.html")
class DescriptorDownloadView(AccessRequiredView):
class DescriptorDownloadView(LoginRequiredMixin, SAMLAccessMixin, View):
"""Replies with the XML Metadata IDSSODescriptor."""
@staticmethod
def get_metadata(request: HttpRequest, provider: SAMLProvider) -> str:
"""Return rendered XML Metadata"""
entity_id = provider.issuer
slo_url = request.build_absolute_uri(
saml_sso_binding_post = request.build_absolute_uri(
reverse(
"passbook_providers_saml:saml-logout",
kwargs={"application": provider.application.slug},
"passbook_providers_saml:sso-post",
kwargs={"application_slug": provider.application.slug},
)
)
sso_post_url = request.build_absolute_uri(
saml_sso_binding_redirect = request.build_absolute_uri(
reverse(
"passbook_providers_saml:saml-login",
kwargs={"application": provider.application.slug},
"passbook_providers_saml:sso-redirect",
kwargs={"application_slug": provider.application.slug},
)
)
subject_format = provider.processor.subject_format
ctx = {
"saml_sso_binding_post": saml_sso_binding_post,
"saml_sso_binding_redirect": saml_sso_binding_redirect,
"entity_id": entity_id,
"slo_url": slo_url,
# Currently, the same endpoint accepts POST and REDIRECT
"sso_post_url": sso_post_url,
"sso_redirect_url": sso_post_url,
"subject_format": subject_format,
}
if provider.signing_kp:
@ -289,9 +259,14 @@ class DescriptorDownloadView(AccessRequiredView):
).replace("\n", "")
return render_to_string("saml/xml/metadata.xml", ctx)
# pylint: disable=unused-argument
def get(self, request: HttpRequest, application: str) -> HttpResponse:
def get(self, request: HttpRequest, application_slug: str) -> HttpResponse:
"""Replies with the XML Metadata IDSSODescriptor."""
self.application = get_object_or_404(Application, slug=application_slug)
self.provider: SAMLProvider = get_object_or_404(
SAMLProvider, pk=self.application.provider_id
)
if not self._has_access():
raise PermissionDenied()
try:
metadata = DescriptorDownloadView.get_metadata(request, self.provider)
except Provider.application.RelatedObjectDoesNotExist: # pylint: disable=no-member

1
passbook/root/settings.py
View file

@ -97,6 +97,7 @@ INSTALLED_APPS = [
"passbook.sources.oauth.apps.PassbookSourceOAuthConfig",
"passbook.sources.saml.apps.PassbookSourceSAMLConfig",
"passbook.stages.captcha.apps.PassbookStageCaptchaConfig",
"passbook.stages.consent.apps.PassbookStageConsentConfig",
"passbook.stages.dummy.apps.PassbookStageDummyConfig",
"passbook.stages.email.apps.PassbookStageEmailConfig",
"passbook.stages.prompt.apps.PassbookStagPromptConfig",

31
passbook/sources/ldap/migrations/0004_auto_20200524_1146.py Normal file
View file

@ -0,0 +1,31 @@
# Generated by Django 3.0.6 on 2020-05-24 11:46
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("passbook_sources_ldap", "0003_default_ldap_property_mappings"),
]
operations = [
migrations.AlterField(
model_name="ldapsource",
name="additional_group_dn",
field=models.TextField(
blank=True,
help_text="Prepended to Base DN for Group-queries.",
verbose_name="Addition Group DN",
),
),
migrations.AlterField(
model_name="ldapsource",
name="additional_user_dn",
field=models.TextField(
blank=True,
help_text="Prepended to Base DN for User-queries.",
verbose_name="Addition User DN",
),
),
]

8
passbook/sources/oauth/forms.py
View file

@ -3,6 +3,7 @@
from django import forms
from passbook.admin.forms.source import SOURCE_FORM_FIELDS
from passbook.flows.models import Flow, FlowDesignation
from passbook.sources.oauth.models import OAuthSource
from passbook.sources.oauth.types.manager import MANAGER
@ -10,6 +11,13 @@ from passbook.sources.oauth.types.manager import MANAGER
class OAuthSourceForm(forms.ModelForm):
"""OAuthSource Form"""
authentication_flow = forms.ModelChoiceField(
queryset=Flow.objects.filter(designation=FlowDesignation.AUTHENTICATION)
)
enrollment_flow = forms.ModelChoiceField(
queryset=Flow.objects.filter(designation=FlowDesignation.ENROLLMENT)
)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if hasattr(self.Meta, "overrides"):

38
passbook/sources/oauth/tests.py Normal file
View file

@ -0,0 +1,38 @@
"""OAuth Source tests"""
from django.shortcuts import reverse
from django.test import Client, TestCase
from passbook.sources.oauth.models import OAuthSource
class OAuthSourceTests(TestCase):
"""OAuth Source tests"""
def setUp(self):
self.client = Client()
self.source = OAuthSource.objects.create(
name="test",
slug="test",
provider_type="openid-connect",
authorization_url="",
profile_url="",
consumer_key="",
)
def test_source_redirect(self):
"""test redirect view"""
self.client.get(
reverse(
"passbook_sources_oauth:oauth-client-login",
kwargs={"source_slug": self.source.slug},
)
)
def test_source_callback(self):
"""test callback view"""
self.client.get(
reverse(
"passbook_sources_oauth:oauth-client-callback",
kwargs={"source_slug": self.source.slug},
)
)

14
passbook/sources/oauth/types/azure_ad.py
View file

@ -1,6 +1,9 @@
"""AzureAD OAuth2 Views"""
import uuid
from typing import Any, Dict
from passbook.core.models import User
from passbook.sources.oauth.models import OAuthSource, UserOAuthSourceConnection
from passbook.sources.oauth.types.manager import MANAGER, RequestKind
from passbook.sources.oauth.utils import user_get_or_create
from passbook.sources.oauth.views.core import OAuthCallback
@ -10,10 +13,15 @@ from passbook.sources.oauth.views.core import OAuthCallback
class AzureADOAuthCallback(OAuthCallback):
"""AzureAD OAuth2 Callback"""
def get_user_id(self, source, info):
return uuid.UUID(info.get("objectId")).int
def get_user_id(self, source: OAuthSource, info: Dict[str, Any]) -> str:
return str(uuid.UUID(info.get("objectId")).int)
def get_or_create_user(self, source, access, info):
def get_or_create_user(
self,
source: OAuthSource,
access: UserOAuthSourceConnection,
info: Dict[str, Any],
) -> User:
user_data = {
"username": info.get("displayName"),
"email": info.get("mail", None) or info.get("otherMails")[0],

4
passbook/sources/oauth/types/manager.py
View file

@ -54,7 +54,9 @@ class SourceTypeManager:
return OAuthCallback
if kind.value == RequestKind.redirect:
return OAuthRedirect
raise KeyError
raise KeyError(
f"Provider Type {source.provider_type} (type {kind.value}) not found."
)
MANAGER = SourceTypeManager()

4
passbook/sources/oauth/types/oidc.py
View file

@ -21,8 +21,8 @@ class OpenIDConnectOAuthRedirect(OAuthRedirect):
class OpenIDConnectOAuth2Callback(OAuthCallback):
"""OpenIDConnect OAuth2 Callback"""
def get_user_id(self, source: OAuthSource, info: Dict[str, str]):
return info.get("sub")
def get_user_id(self, source: OAuthSource, info: Dict[str, str]) -> str:
return info.get("sub", "")
def get_or_create_user(self, source: OAuthSource, access, info: Dict[str, str]):
user_data = {

77
passbook/sources/oauth/views/core.py
View file

@ -1,11 +1,11 @@
"""Core OAauth Views"""
from typing import Callable, Optional
from typing import Any, Callable, Dict, Optional
from django.conf import settings
from django.contrib import messages
from django.contrib.auth import authenticate
from django.contrib.auth.mixins import LoginRequiredMixin
from django.http import Http404
from django.http import Http404, HttpRequest, HttpResponse
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
from django.utils.translation import ugettext as _
@ -13,7 +13,8 @@ from django.views.generic import RedirectView, View
from structlog import get_logger
from passbook.audit.models import Event, EventAction
from passbook.flows.models import Flow, FlowDesignation
from passbook.core.models import User
from passbook.flows.models import Flow
from passbook.flows.planner import (
PLAN_CONTEXT_PENDING_USER,
PLAN_CONTEXT_SSO,
@ -49,18 +50,18 @@ class OAuthRedirect(OAuthClientMixin, RedirectView):
params = None
# pylint: disable=unused-argument
def get_additional_parameters(self, source):
def get_additional_parameters(self, source: OAuthSource) -> Dict[str, Any]:
"Return additional redirect parameters for this source."
return self.params or {}
def get_callback_url(self, source):
def get_callback_url(self, source: OAuthSource) -> str:
"Return the callback url for this source."
return reverse(
"passbook_sources_oauth:oauth-client-callback",
kwargs={"source_slug": source.slug},
)
def get_redirect_url(self, **kwargs):
def get_redirect_url(self, **kwargs) -> str:
"Build redirect url for a given source."
slug = kwargs.get("source_slug", "")
try:
@ -84,7 +85,7 @@ class OAuthCallback(OAuthClientMixin, View):
source_id = None
source = None
def get(self, request, *_, **kwargs):
def get(self, request: HttpRequest, *_, **kwargs) -> HttpResponse:
"""View Get handler"""
slug = kwargs.get("source_slug", "")
try:
@ -143,38 +144,38 @@ class OAuthCallback(OAuthClientMixin, View):
return self.handle_existing_user(self.source, user, connection, info)
# pylint: disable=unused-argument
def get_callback_url(self, source):
def get_callback_url(self, source: OAuthSource) -> str:
"Return callback url if different than the current url."
return False
return ""
# pylint: disable=unused-argument
def get_error_redirect(self, source, reason):
def get_error_redirect(self, source: OAuthSource, reason: str) -> str:
"Return url to redirect on login failure."
return settings.LOGIN_URL
def get_or_create_user(self, source, access, info):
def get_or_create_user(
self,
source: OAuthSource,
access: UserOAuthSourceConnection,
info: Dict[str, Any],
) -> User:
"Create a shell auth.User."
raise NotImplementedError()
# pylint: disable=unused-argument
def get_user_id(self, source, info):
"Return unique identifier from the profile info."
id_key = self.source_id or "id"
result = info
try:
for key in id_key.split("."):
result = result[key]
return result
except KeyError:
def get_user_id(
self, source: UserOAuthSourceConnection, info: Dict[str, Any]
) -> Optional[str]:
"""Return unique identifier from the profile info."""
if "id" in info:
return info["id"]
return None
def handle_login(self, user, source, access):
def handle_login_flow(self, flow: Optional[Flow], user: User) -> HttpResponse:
"""Prepare Authentication Plan, redirect user FlowExecutor"""
user = authenticate(
source=access.source, identifier=access.identifier, request=self.request
)
if not flow:
raise Http404
# We run the Flow planner here so we can pass the Pending user in the context
flow = get_object_or_404(Flow, designation=FlowDesignation.AUTHENTICATION)
planner = FlowPlanner(flow)
plan = planner.plan(
self.request,
@ -186,11 +187,17 @@ class OAuthCallback(OAuthClientMixin, View):
)
self.request.session[SESSION_KEY_PLAN] = plan
return redirect_with_qs(
"passbook_flows:flow-executor", self.request.GET, flow_slug=flow.slug,
"passbook_flows:flow-executor-shell", self.request.GET, flow_slug=flow.slug,
)
# pylint: disable=unused-argument
def handle_existing_user(self, source, user, access, info):
def handle_existing_user(
self,
source: OAuthSource,
user: User,
access: UserOAuthSourceConnection,
info: Dict[str, Any],
) -> HttpResponse:
"Login user and redirect."
messages.success(
self.request,
@ -199,15 +206,23 @@ class OAuthCallback(OAuthClientMixin, View):
% {"source": self.source.name}
),
)
return self.handle_login(user, source, access)
user = authenticate(
source=access.source, identifier=access.identifier, request=self.request
)
return self.handle_login_flow(source.authentication_flow, user)
def handle_login_failure(self, source, reason):
def handle_login_failure(self, source: OAuthSource, reason: str) -> HttpResponse:
"Message user and redirect on error."
LOGGER.warning("Authentication Failure", reason=reason)
messages.error(self.request, _("Authentication Failed."))
return redirect(self.get_error_redirect(source, reason))
def handle_new_user(self, source, access, info):
def handle_new_user(
self,
source: OAuthSource,
access: UserOAuthSourceConnection,
info: Dict[str, Any],
) -> HttpResponse:
"Create a shell auth.User and redirect."
was_authenticated = False
if self.request.user.is_authenticated:
@ -244,7 +259,7 @@ class OAuthCallback(OAuthClientMixin, View):
% {"source": self.source.name}
),
)
return self.handle_login(user, source, access)
return self.handle_login_flow(source.enrollment_flow, user)
class DisconnectView(LoginRequiredMixin, View):

1
passbook/sources/saml/forms.py
View file

@ -28,3 +28,4 @@ class SAMLSourceForm(forms.ModelForm):
"idp_url": forms.TextInput(),
"idp_logout_url": forms.TextInput(),
}
labels = {"signing_kp": _("Singing Keypair")}

30
passbook/sources/saml/migrations/0002_auto_20200523_2329.py Normal file
View file

@ -0,0 +1,30 @@
# Generated by Django 3.0.6 on 2020-05-23 23:29
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("passbook_sources_saml", "0001_initial"),
]
operations = [
migrations.AddField(
model_name="samlsource",
name="binding_type",
field=models.CharField(
choices=[("REDIRECT", "Redirect"), ("POST", "Post")],
default="REDIRECT",
max_length=100,
),
),
migrations.AlterField(
model_name="samlsource",
name="idp_url",
field=models.URLField(
help_text="URL that the initial SAML Request is sent to. Also known as a Binding.",
verbose_name="IDP URL",
),
),
]

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

@ -8,6 +8,13 @@ from passbook.core.types import UILoginButton
from passbook.crypto.models import CertificateKeyPair
class SAMLBindingTypes(models.TextChoices):
"""SAML Binding types"""
Redirect = "REDIRECT"
POST = "POST"
class SAMLSource(Source):
"""SAML Source"""
@ -18,7 +25,18 @@ class SAMLSource(Source):
help_text=_("Also known as Entity ID. Defaults the Metadata URL."),
)
idp_url = models.URLField(verbose_name=_("IDP URL"))
idp_url = models.URLField(
verbose_name=_("IDP URL"),
help_text=_(
"URL that the initial SAML Request is sent to. Also known as a Binding."
),
)
binding_type = models.CharField(
max_length=100,
choices=SAMLBindingTypes.choices,
default=SAMLBindingTypes.Redirect,
)
idp_logout_url = models.URLField(
default=None, blank=True, null=True, verbose_name=_("IDP Logout URL")
)

58
passbook/sources/saml/processors/base.py
View file

@ -2,21 +2,31 @@
from typing import TYPE_CHECKING, Optional
from defusedxml import ElementTree
from django.http import HttpRequest
from django.http import HttpRequest, HttpResponse
from signxml import XMLVerifier
from structlog import get_logger
from passbook.core.models import User
from passbook.flows.planner import (
PLAN_CONTEXT_PENDING_USER,
PLAN_CONTEXT_SSO,
FlowPlanner,
)
from passbook.flows.views import SESSION_KEY_PLAN
from passbook.lib.utils.urls import redirect_with_qs
from passbook.providers.saml.utils.encoding import decode_base64_and_inflate
from passbook.sources.saml.exceptions import (
MissingSAMLResponse,
UnsupportedNameIDFormat,
)
from passbook.sources.saml.models import SAMLSource
from passbook.stages.password.stage import PLAN_CONTEXT_AUTHENTICATION_BACKEND
from passbook.stages.prompt.stage import PLAN_CONTEXT_PROMPT
LOGGER = get_logger()
if TYPE_CHECKING:
from xml.etree.ElementTree import Element # nosec
DEFAULT_BACKEND = "django.contrib.auth.backends.ModelBackend"
class Processor:
@ -46,7 +56,9 @@ class Processor:
def _verify_signed(self):
"""Verify SAML Response's Signature"""
verifier = XMLVerifier()
verifier.verify(self._root_xml, x509_cert=self._source.signing_kp.certificate)
verifier.verify(
self._root_xml, x509_cert=self._source.signing_kp.certificate_data
)
def _get_email(self) -> Optional[str]:
"""
@ -69,18 +81,32 @@ class Processor:
)
return name_id.text
def get_user(self) -> User:
"""
Gets info out of the response and locally logs in this user.
May create a local user account first.
Returns the user object that was created.
"""
def prepare_flow(self, request: HttpRequest) -> HttpResponse:
"""Prepare flow plan depending on whether or not the user exists"""
email = self._get_email()
try:
user = User.objects.get(email=email)
except User.DoesNotExist:
user = User.objects.create_user(username=email, email=email)
# TODO: Property Mappings
user.set_unusable_password()
user.save()
return user
matching_users = User.objects.filter(email=email)
if matching_users.exists():
# User exists already, switch to authentication flow
flow = self._source.authentication_flow
request.session[SESSION_KEY_PLAN] = FlowPlanner(flow).plan(
request,
{
# Data for authentication
PLAN_CONTEXT_PENDING_USER: matching_users.first(),
PLAN_CONTEXT_AUTHENTICATION_BACKEND: DEFAULT_BACKEND,
PLAN_CONTEXT_SSO: True,
},
)
else:
flow = self._source.enrollment_flow
request.session[SESSION_KEY_PLAN] = FlowPlanner(flow).plan(
request,
{
# Data for enrollment
PLAN_CONTEXT_PROMPT: {"username": email, "email": email},
PLAN_CONTEXT_SSO: True,
},
)
return redirect_with_qs(
"passbook_flows:flow-executor-shell", request.GET, flow_slug=flow.slug,
)

15
passbook/sources/saml/views.py
View file

@ -1,8 +1,9 @@
"""saml sp views"""
from django.contrib.auth import login, logout
from django.contrib.auth import logout
from django.http import Http404, HttpRequest, HttpResponse
from django.shortcuts import get_object_or_404, redirect, render, reverse
from django.shortcuts import get_object_or_404, redirect, render
from django.utils.decorators import method_decorator
from django.utils.http import urlencode
from django.views import View
from django.views.decorators.csrf import csrf_exempt
from signxml.util import strip_pem_header
@ -15,7 +16,7 @@ from passbook.sources.saml.exceptions import (
MissingSAMLResponse,
UnsupportedNameIDFormat,
)
from passbook.sources.saml.models import SAMLSource
from passbook.sources.saml.models import SAMLBindingTypes, SAMLSource
from passbook.sources.saml.processors.base import Processor
from passbook.sources.saml.utils import build_full_url, get_issuer
from passbook.sources.saml.xml_render import get_authnrequest_xml
@ -40,6 +41,9 @@ class InitiateView(View):
}
authn_req = get_authnrequest_xml(parameters, signed=False)
_request = nice64(str.encode(authn_req))
if source.binding_type == SAMLBindingTypes.Redirect:
return redirect(source.idp_url + "?" + urlencode({"SAMLRequest": _request}))
if source.binding_type == SAMLBindingTypes.POST:
return render(
request,
"saml/sp/login.html",
@ -50,6 +54,7 @@ class InitiateView(View):
"source": source,
},
)
raise Http404
@method_decorator(csrf_exempt, name="dispatch")
@ -68,9 +73,7 @@ class ACSView(View):
return bad_request_message(request, str(exc))
try:
user = processor.get_user()
login(request, user, backend="django.contrib.auth.backends.ModelBackend")
return redirect(reverse("passbook_core:overview"))
return processor.prepare_flow(request)
except UnsupportedNameIDFormat as exc:
return bad_request_message(request, str(exc))

8
passbook/stages/captcha/tests.py
View file

@ -2,6 +2,7 @@
from django.conf import settings
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.flows.models import Flow, FlowDesignation, FlowStageBinding
@ -44,5 +45,8 @@ class TestCaptchaStage(TestCase):
),
{"g-recaptcha-response": "PASSED"},
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_core:overview"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)

0
passbook/stages/consent/__init__.py Normal file
View file

21
passbook/stages/consent/api.py Normal file
View file

@ -0,0 +1,21 @@
"""ConsentStage API Views"""
from rest_framework.serializers import ModelSerializer
from rest_framework.viewsets import ModelViewSet
from passbook.stages.consent.models import ConsentStage
class ConsentStageSerializer(ModelSerializer):
"""ConsentStage Serializer"""
class Meta:
model = ConsentStage
fields = ["pk", "name"]
class ConsentStageViewSet(ModelViewSet):
"""ConsentStage Viewset"""
queryset = ConsentStage.objects.all()
serializer_class = ConsentStageSerializer

10
passbook/stages/consent/apps.py Normal file
View file

@ -0,0 +1,10 @@
"""passbook consent app"""
from django.apps import AppConfig
class PassbookStageConsentConfig(AppConfig):
"""passbook consent app"""
name = "passbook.stages.consent"
label = "passbook_stages_consent"
verbose_name = "passbook Stages.Consent"

20
passbook/stages/consent/forms.py Normal file
View file

@ -0,0 +1,20 @@
"""passbook consent stage forms"""
from django import forms
from passbook.stages.consent.models import ConsentStage
class ConsentForm(forms.Form):
"""passbook consent stage form"""
class ConsentStageForm(forms.ModelForm):
"""Form to edit ConsentStage Instance"""
class Meta:
model = ConsentStage
fields = ["name"]
widgets = {
"name": forms.TextInput(),
}

37
passbook/stages/consent/migrations/0001_initial.py Normal file
View file

@ -0,0 +1,37 @@
# Generated by Django 3.0.6 on 2020-05-24 11:46
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
("passbook_flows", "0004_source_flows"),
]
operations = [
migrations.CreateModel(
name="ConsentStage",
fields=[
(
"stage_ptr",
models.OneToOneField(
auto_created=True,
on_delete=django.db.models.deletion.CASCADE,
parent_link=True,
primary_key=True,
serialize=False,
to="passbook_flows.Stage",
),
),
],
options={
"verbose_name": "Consent Stage",
"verbose_name_plural": "Consent Stages",
},
bases=("passbook_flows.stage",),
),
]

0
passbook/stages/consent/migrations/__init__.py Normal file
View file

19
passbook/stages/consent/models.py Normal file
View file

@ -0,0 +1,19 @@
"""passbook consent stage"""
from django.utils.translation import gettext_lazy as _
from passbook.flows.models import Stage
class ConsentStage(Stage):
"""Consent Stage instance"""
type = "passbook.stages.consent.stage.ConsentStage"
form = "passbook.stages.consent.forms.ConsentStageForm"
def __str__(self):
return f"Consent Stage {self.name}"
class Meta:
verbose_name = _("Consent Stage")
verbose_name_plural = _("Consent Stages")

25
passbook/stages/consent/stage.py Normal file
View file

@ -0,0 +1,25 @@
"""passbook consent stage"""
from typing import Any, Dict
from django.views.generic import FormView
from passbook.flows.stage import StageView
from passbook.lib.utils.template import render_to_string
from passbook.stages.consent.forms import ConsentForm
class ConsentStage(FormView, StageView):
"""Simple consent checker."""
body_template_name: str
form_class = ConsentForm
def get_context_data(self, **kwargs: Dict[str, Any]) -> Dict[str, Any]:
kwargs = super().get_context_data(**kwargs)
if self.body_template_name:
kwargs["body"] = render_to_string(self.body_template_name, kwargs)
return kwargs
def form_valid(self, form):
return self.executor.stage_ok()

47
passbook/stages/consent/tests.py Normal file
View file

@ -0,0 +1,47 @@
"""consent tests"""
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.flows.models import Flow, FlowDesignation, FlowStageBinding
from passbook.flows.planner import FlowPlan
from passbook.flows.views import SESSION_KEY_PLAN
from passbook.stages.consent.models import ConsentStage
class TestConsentStage(TestCase):
"""Consent tests"""
def setUp(self):
super().setUp()
self.user = User.objects.create_user(
username="unittest", email="test@beryju.org"
)
self.client = Client()
self.flow = Flow.objects.create(
name="test-consent",
slug="test-consent",
designation=FlowDesignation.AUTHENTICATION,
)
self.stage = ConsentStage.objects.create(name="consent",)
FlowStageBinding.objects.create(flow=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])
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}
),
{},
)
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)

8
passbook/stages/dummy/tests.py
View file

@ -1,6 +1,7 @@
"""dummy tests"""
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.flows.models import Flow, FlowDesignation, FlowStageBinding
@ -41,8 +42,11 @@ class TestDummyStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
response = self.client.post(url, {})
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_core:overview"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
def test_form(self):
"""Test Form"""

9
passbook/stages/email/tests.py
View file

@ -4,6 +4,7 @@ from unittest.mock import MagicMock, patch
from django.core import mail
from django.shortcuts import reverse
from django.test import Client, TestCase
from django.utils.encoding import force_text
from passbook.core.models import Token, User
from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
@ -93,8 +94,12 @@ class TestEmailStage(TestCase):
token = Token.objects.get(user=self.user)
url += f"?{QS_KEY_TOKEN}={token.pk.hex}"
response = self.client.get(url)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_core:overview"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
session = self.client.session
plan: FlowPlan = session[SESSION_KEY_PLAN]

12
passbook/stages/identification/tests.py
View file

@ -1,6 +1,7 @@
"""identification tests"""
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.flows.models import Flow, FlowDesignation, FlowStageBinding
@ -53,8 +54,11 @@ class TestIdentificationStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
response = self.client.post(url, form_data)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_core:overview"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
def test_invalid_with_username(self):
"""Test invalid with username (user exists but stage only allows e-mail)"""
@ -97,7 +101,7 @@ class TestIdentificationStage(TestCase):
),
)
self.assertEqual(response.status_code, 200)
self.assertIn(flow.slug, response.rendered_content)
self.assertIn(flow.slug, force_text(response.content))
def test_recovery_flow(self):
"""Test that recovery flow is linked correctly"""
@ -118,4 +122,4 @@ class TestIdentificationStage(TestCase):
),
)
self.assertEqual(response.status_code, 200)
self.assertIn(flow.slug, response.rendered_content)
self.assertIn(flow.slug, force_text(response.content))

25
passbook/stages/invitation/tests.py
View file

@ -3,6 +3,7 @@ from unittest.mock import MagicMock, patch
from django.shortcuts import reverse
from django.test import Client, TestCase
from django.utils.encoding import force_text
from guardian.shortcuts import get_anonymous_user
from passbook.core.models import User
@ -52,8 +53,12 @@ class TestUserLoginStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_flows:denied"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_flows:denied")},
)
def test_without_invitation_continue(self):
"""Test without any invitation, continue_flow_without_invitation is set."""
@ -73,8 +78,13 @@ class TestUserLoginStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_core:overview"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
self.stage.continue_flow_without_invitation = False
self.stage.save()
@ -106,5 +116,8 @@ class TestUserLoginStage(TestCase):
plan: FlowPlan = session[SESSION_KEY_PLAN]
self.assertEqual(plan.context[PLAN_CONTEXT_PROMPT], data)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_core:overview"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)

27
passbook/stages/password/tests.py
View file

@ -6,6 +6,7 @@ from unittest.mock import MagicMock, patch
from django.core.exceptions import PermissionDenied
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.flows.models import Flow, FlowDesignation, FlowStageBinding
@ -54,8 +55,12 @@ class TestPasswordStage(TestCase):
# Still have to send the password so the form is valid
{"password": self.password},
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_flows:denied"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_flows:denied")},
)
def test_recovery_flow_link(self):
"""Test link to the default recovery flow"""
@ -74,7 +79,7 @@ class TestPasswordStage(TestCase):
),
)
self.assertEqual(response.status_code, 200)
self.assertIn(flow.slug, response.rendered_content)
self.assertIn(flow.slug, force_text(response.content))
def test_valid_password(self):
"""Test with a valid pending user and valid password"""
@ -91,8 +96,12 @@ class TestPasswordStage(TestCase):
# Form data
{"password": self.password},
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_core:overview"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
def test_invalid_password(self):
"""Test with a valid pending user and invalid password"""
@ -131,5 +140,9 @@ class TestPasswordStage(TestCase):
# Form data
{"password": self.password + "test"},
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_flows:denied"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_flows:denied")},
)

14
passbook/stages/prompt/tests.py
View file

@ -3,6 +3,7 @@ from unittest.mock import MagicMock, patch
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.flows.models import Flow, FlowDesignation, FlowStageBinding
@ -107,9 +108,9 @@ class TestPromptStage(TestCase):
)
self.assertEqual(response.status_code, 200)
for prompt in self.stage.fields.all():
self.assertIn(prompt.field_key, response.rendered_content)
self.assertIn(prompt.label, response.rendered_content)
self.assertIn(prompt.placeholder, response.rendered_content)
self.assertIn(prompt.field_key, force_text(response.content))
self.assertIn(prompt.label, force_text(response.content))
self.assertIn(prompt.placeholder, force_text(response.content))
def test_valid_form_with_policy(self) -> PromptForm:
"""Test form validation"""
@ -151,8 +152,11 @@ class TestPromptStage(TestCase):
),
form.cleaned_data,
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_core:overview"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
# Check that valid data has been saved
session = self.client.session

15
passbook/stages/user_delete/tests.py
View file

@ -1,6 +1,7 @@
"""delete tests"""
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.flows.models import Flow, FlowDesignation, FlowStageBinding
@ -38,8 +39,11 @@ class TestUserDeleteStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_flows:denied"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_flows:denied")},
)
def test_user_delete_get(self):
"""Test Form render"""
@ -70,5 +74,10 @@ class TestUserDeleteStage(TestCase):
),
{},
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
self.assertFalse(User.objects.filter(username=self.username).exists())

25
passbook/stages/user_login/tests.py
View file

@ -1,6 +1,7 @@
"""login tests"""
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.flows.models import Flow, FlowDesignation, FlowStageBinding
@ -43,8 +44,12 @@ class TestUserLoginStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_core:overview"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
def test_without_user(self):
"""Test a plan without any pending user, resulting in a denied"""
@ -58,8 +63,12 @@ class TestUserLoginStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_flows:denied"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_flows:denied")},
)
def test_without_backend(self):
"""Test a plan with pending user, without backend, resulting in a denied"""
@ -74,8 +83,12 @@ class TestUserLoginStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_flows:denied"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_flows:denied")},
)
def test_form(self):
"""Test Form"""

9
passbook/stages/user_logout/tests.py
View file

@ -1,6 +1,7 @@
"""logout tests"""
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.flows.models import Flow, FlowDesignation, FlowStageBinding
@ -43,8 +44,12 @@ class TestUserLogoutStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_core:overview"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
def test_form(self):
"""Test Form"""

23
passbook/stages/user_write/tests.py
View file

@ -4,6 +4,7 @@ from random import SystemRandom
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.flows.models import Flow, FlowDesignation, FlowStageBinding
@ -52,7 +53,12 @@ class TestUserWriteStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
user_qs = User.objects.filter(
username=plan.context[PLAN_CONTEXT_PROMPT]["username"]
)
@ -83,7 +89,12 @@ class TestUserWriteStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
user_qs = User.objects.filter(
username=plan.context[PLAN_CONTEXT_PROMPT]["username"]
)
@ -103,8 +114,12 @@ class TestUserWriteStage(TestCase):
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_flows:denied"))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
force_text(response.content),
{"type": "redirect", "to": reverse("passbook_flows:denied")},
)
def test_form(self):
"""Test Form"""

2
pyproject.toml Normal file
View file

@ -0,0 +1,2 @@
[tool.black]
target-version = ['py37']

47
swagger.yaml
View file

@ -4875,9 +4875,6 @@ definitions:
pattern: ^[-a-zA-Z0-9_]+$
maxLength: 50
minLength: 1
skip_authorization:
title: Skip authorization
type: boolean
provider:
title: Provider
type: integer
@ -5048,6 +5045,7 @@ definitions:
type: string
enum:
- authentication
- authorization
- invalidation
- enrollment
- unenrollment
@ -5335,12 +5333,19 @@ definitions:
type: string
minLength: 1
Provider:
required:
- authorization_flow
type: object
properties:
pk:
title: ID
type: integer
readOnly: true
authorization_flow:
title: Authorization flow
description: Flow used when authorizing this provider.
type: string
format: uuid
property_mappings:
type: array
items:
@ -5594,6 +5599,18 @@ definitions:
enabled:
title: Enabled
type: boolean
authentication_flow:
title: Authentication flow
description: Flow to use when authenticating existing users.
type: string
format: uuid
x-nullable: true
enrollment_flow:
title: Enrollment flow
description: Flow to use when enrolling new users.
type: string
format: uuid
x-nullable: true
__type__:
title: 'type '
type: string
@ -5629,6 +5646,18 @@ definitions:
enabled:
title: Enabled
type: boolean
authentication_flow:
title: Authentication flow
description: Flow to use when authenticating existing users.
type: string
format: uuid
x-nullable: true
enrollment_flow:
title: Enrollment flow
description: Flow to use when enrolling new users.
type: string
format: uuid
x-nullable: true
server_uri:
title: Server URI
type: string
@ -5726,6 +5755,18 @@ definitions:
enabled:
title: Enabled
type: boolean
authentication_flow:
title: Authentication flow
description: Flow to use when authenticating existing users.
type: string
format: uuid
x-nullable: true
enrollment_flow:
title: Enrollment flow
description: Flow to use when enrolling new users.
type: string
format: uuid
x-nullable: true
provider_type:
title: Provider type
type: string