debian/debian_config_common.py

@@ -269,6 +269,7 @@ if 'MIDDLEWARE' not in globals():
     MIDDLEWARE = global_settings.MIDDLEWARE
 
 MIDDLEWARE = (
+    'hobo.middleware.security.content_security_policy_middleware',
     'hobo.middleware.VersionMiddleware',  # /__version__
     'hobo.middleware.cors.CORSMiddleware',
     'hobo.middleware.maintenance.MaintenanceMiddleware',

hobo/environment/models.py

@@ -101,6 +101,22 @@ class Variable(models.Model):
     def clean(self):
         self._parse_value_as_json()
 
+    def update_value(self, value):
+        if self.value != value:
+            self.value = value
+            self.save()
+            return True
+        else:
+            return False
+
+    def update_json(self, value):
+        if self.json != value:
+            self.json = value
+            self.save()
+            return True
+        else:
+            return False
+
 
 def is_resolvable(url):
     try:

hobo/maintenance/views.py

@@ -68,21 +68,16 @@ class HomeView(FormView):
             form.add_confirmation_checkbox()
             return self.form_invalid(form)
 
-        self.maintenance_page_variable.json = form.cleaned_data['maintenance_page']
-        self.maintenance_page_variable.save()
-        self.maintenance_page_message_variable.value = form.cleaned_data['maintenance_page_message']
-        self.maintenance_page_message_variable.save()
+        self.maintenance_page_variable.update_json(form.cleaned_data['maintenance_page'])
+        self.maintenance_page_message_variable.update_value(form.cleaned_data['maintenance_page_message'])
+        self.maintenance_pass_through_header_variable.update_value(
+            form.cleaned_data['maintenance_pass_through_header']
+        )
+        self.maintenance_pass_through_dnswl_variable.update_value(
+            form.cleaned_data['maintenance_pass_through_dnswl']
+        )
+        self.tenant_disable_cron_jobs_variable.update_json(form.cleaned_data['disable_cron'])
 
-        self.maintenance_pass_through_header_variable.value = form.cleaned_data[
-            'maintenance_pass_through_header'
-        ]
-        self.maintenance_pass_through_header_variable.save()
-        self.maintenance_pass_through_dnswl_variable.value = form.cleaned_data[
-            'maintenance_pass_through_dnswl'
-        ]
-        self.maintenance_pass_through_dnswl_variable.save()
-        self.tenant_disable_cron_jobs_variable.json = form.cleaned_data['disable_cron']
-        self.tenant_disable_cron_jobs_variable.save()
         return super().form_valid(form)
 
     def get_context_data(self, **kwargs):

hobo/middleware/security.py

@@ -0,0 +1,51 @@
+# hobo - portal to configure and deploy applications
+# Copyright (C) Entr'ouvert
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+from django.conf import settings
+
+CSP_HEADER_NAME = 'Content-Security-Policy'
+CSP_REPORT_ONLY_HEADER_NAME = 'Content-Security-Policy-Report-Only'
+
+
+def content_security_policy_middleware(get_response):
+    def middleware(request):
+        response = get_response(request)
+        content_security_policy = getattr(settings, 'CONTENT_SECURITY_POLICY', None)
+        content_security_policy_report_only = getattr(settings, 'CONTENT_SECURITY_POLICY_REPORT_ONLY', None)
+        content_security_policy_report_uri = getattr(settings, 'CONTENT_SECURITY_POLICY_REPORT_URI', None)
+
+        if content_security_policy and CSP_HEADER_NAME not in response.headers:
+            response[CSP_HEADER_NAME] = content_security_policy
+
+        if content_security_policy_report_only and CSP_REPORT_ONLY_HEADER_NAME not in response.headers:
+            response[CSP_REPORT_ONLY_HEADER_NAME] = content_security_policy_report_only
+
+        if content_security_policy_report_uri:
+            if policy := response.headers.get(CSP_HEADER_NAME):
+                if 'report-uri' not in policy:
+                    response[CSP_HEADER_NAME] = (
+                        policy + f'; report-uri {content_security_policy_report_uri}?error'
+                    )
+
+            if policy := response.headers.get(CSP_REPORT_ONLY_HEADER_NAME):
+                if 'report-uri' not in policy:
+                    response[CSP_REPORT_ONLY_HEADER_NAME] = (
+                        policy + f'; report-uri {content_security_policy_report_uri}'
+                    )
+
+        return response
+
+    return middleware

hobo/security/forms.py

@@ -0,0 +1,98 @@
+# hobo - portal to configure and deploy applications
+# Copyright (C) 2015-2022  Entr'ouvert
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import re
+
+from django import forms
+from django.core.exceptions import ValidationError
+from django.utils.translation import gettext_lazy as _
+
+HEADS = [
+    'child-src',
+    'default-src',
+    'font-src',
+    'frame-src',
+    'img-src',
+    'manifest-src',
+    'media-src',
+    'prefetch-src',
+    'script-src',
+    'style-src',
+    'worker-src',
+    'form-action',
+    'frame-ancestors',
+    'navigate-to',
+    'report-uri',
+]
+
+SOURCES = [
+    "'self'",
+    "'unsafe-eval'",
+    "'unsafe-inline'",
+    "'none'",
+    'data:',
+    re.compile(r'^\*(:?\.[0-9a-z-]+)*$'),
+    re.compile(r'^[0-9a-z-]+(:?\.[0-9a-z-]+)+$'),
+    re.compile(r'^https://\*(:?\.[0-9a-z-]+)*$'),
+    re.compile(r'^https://[0-9a-z-]+(:?\.[0-9a-z-]+)+$'),
+    re.compile(r'^https://\*(:?\.[0-9a-z-]+)*:[0-9]+$'),
+    re.compile(r'^https://[0-9a-z-]+(:?\.[0-9a-z-]+)+:[0-9]$'),
+]
+
+
+def validate_csp(policy):
+    '''Validate a subset of possibles CSP policies'''
+
+    policy = policy.replace('\n', ' ')
+    directives = list(filter(None, (directive.strip() for directive in policy.split(';'))))
+    for directive in directives:
+        head, rest = directive.split(' ', 1)
+        if head not in HEADS:
+            raise ValidationError(_('Invalid CSP directive "%s"') % head)
+        sources = rest.split(' ')
+        for source in sources:
+            for needle in SOURCES:
+                if hasattr(needle, 'match'):
+                    if needle.match(source):
+                        break
+                else:
+                    if needle == source:
+                        break
+            else:
+                raise ValidationError(_('Invalid CSP source "%s" in directive "%s"') % (source, head))
+
+
+class SecurityForm(forms.Form):
+    content_security_policy_report_only = forms.CharField(
+        required=False,
+        widget=forms.Textarea,
+        label=_('Content Security Policy Report Only'),
+        help_text=_('You should first test your policy by setting a "Report Only" content security policy'),
+        validators=[validate_csp],
+    )
+
+    content_security_policy = forms.CharField(
+        required=False,
+        widget=forms.Textarea,
+        label=_('Content Security Policy'),
+        validators=[validate_csp],
+    )
+
+    def clean(self):
+        cleaned_data = self.cleaned_data
+        for key, value in cleaned_data.items():
+            cleaned_data[key] = re.sub(r'\s+', ' ', value.strip())
+        return cleaned_data

hobo/security/migrations/0001_initial.py

@@ -0,0 +1,36 @@
+# Generated by Django 3.2.18 on 2024-01-17 13:20
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name='CspReport',
+            fields=[
+                (
+                    'id',
+                    models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
+                ),
+                ('first_seen', models.DateTimeField(auto_now_add=True, verbose_name='First seen')),
+                ('last_seen', models.DateTimeField(auto_now_add=True, verbose_name='Last seen')),
+                ('error', models.BooleanField(verbose_name='Error')),
+                ('inline', models.BooleanField(verbose_name='Inline')),
+                ('source', models.TextField(verbose_name='Source')),
+                ('line', models.PositiveIntegerField(verbose_name='Line')),
+                ('column', models.PositiveIntegerField(verbose_name='Column')),
+                ('violated_directive', models.TextField(verbose_name='Violated directive')),
+                ('content', models.JSONField(verbose_name='Content')),
+            ],
+            options={
+                'verbose_name': 'CSP report',
+                'verbose_name_plural': 'CSP reports',
+                'db_table': 'hobo_cspreport',
+                'unique_together': {('error', 'inline', 'source', 'violated_directive', 'line', 'column')},
+            },
+        ),
+    ]

hobo/security/models.py

@@ -0,0 +1,101 @@
+# hobo - portal to configure and deploy applications
+# Copyright (C) Entr'ouvert
+
+import datetime
+import json
+
+from django.core.cache import cache
+from django.db import models
+from django.utils.timezone import now
+from django.utils.translation import gettext_lazy as _
+
+
+class CspReport(models.Model):
+    first_seen = models.DateTimeField(auto_now_add=True, verbose_name=_('First seen'))
+    last_seen = models.DateTimeField(auto_now_add=True, verbose_name=_('Last seen'))
+    error = models.BooleanField(verbose_name=_('Error'))
+    inline = models.BooleanField(verbose_name=_('Inline'))
+    source = models.TextField(verbose_name=_('Source'))
+    line = models.PositiveIntegerField(verbose_name=_('Line'))
+    column = models.PositiveIntegerField(verbose_name=_('Column'))
+    violated_directive = models.TextField(verbose_name=_('Violated directive'))
+    content = models.JSONField(verbose_name=_('Content'))
+
+    # period to update report.last_seen
+    UPDATE_LAST_SEEN_PERIOD = datetime.timedelta(minutes=10)
+
+    def get_content(self):
+        return {key.replace('-', '_'): value for key, value in self.content.items()}
+
+    def is_source_inline(self):
+        return self.source.startswith('inline ')
+
+    def full(self):
+        return json.dumps(self.content, indent=2)
+
+    @classmethod
+    def record_from_report(cls, report, error=False):
+        '''Create a new CspReport from the deserialized JSON pushed to a report-uri endpoint.'''
+        try:
+            csp_report = report['csp-report']
+            violated_directive = csp_report['violated-directive']
+            blocked_uri = csp_report['blocked-uri']
+            document_uri = csp_report['document-uri']
+            inline = blocked_uri.lower() == 'inline'
+            line = csp_report.get('line-number', 0)
+            column = csp_report.get('column-number', 0)
+            if inline:
+                source = document_uri
+            else:
+                source = blocked_uri
+        except Exception as e:
+            raise ValueError('invalid csp-report: %s' % e)
+
+        # keep one report per policy violation
+        csp_report, created = CspReport.objects.get_or_create(
+            error=error,
+            inline=inline,
+            source=source,
+            violated_directive=violated_directive,
+            line=line,
+            column=column,
+            defaults={'content': csp_report},
+        )
+
+        # update CspReport.last_seen not faster than every 10 minutes...
+        if not created:
+            time_since_last_seen = now() - csp_report.last_seen
+            if time_since_last_seen > cls.UPDATE_LAST_SEEN_PERIOD:
+                CspReport.objects.filter(pk=csp_report.pk).update(last_seen=now())
+
+        return csp_report
+
+    @classmethod
+    def clean_old_reports(cls, **kwargs):
+        not_on_or_after = now() - datetime.timedelta(**kwargs)
+        cls.objects.filter(last_seen__lt=not_on_or_after).delete()
+
+    def __str__(self):
+        if self.inline:
+            return f'blocked inline asset in {self.source}, line {self.line}, column: {self.column} by directive {self.violated_directive}'
+        else:
+            return f'blocked asset {self.source} by directive {self.violated_directive}'
+
+    class Meta:
+        verbose_name = _('CSP report')
+        verbose_name_plural = _('CSP reports')
+        db_table = 'hobo_cspreport'
+        unique_together = [
+            ('error', 'inline', 'source', 'violated_directive', 'line', 'column'),
+        ]
+
+    @classmethod
+    def too_much_reports(cls):
+        '''Verify if too much reports have been reported'''
+        cache_key = 'csp-too-much-reports'
+        cache_value = cache.get(cache_key)
+        if cache_value is None:
+            cls.clean_old_reports(days=1)
+            cache_value = cls.objects.count() > 1000
+            cache.set(cache_key, cache_value, 60)
+        return cache_value

hobo/security/templates/hobo/security/home.html

@@ -0,0 +1,77 @@
+{% extends "hobo/base.html" %}
+{% load i18n %}
+
+{% block breadcrumb %}
+  {{ block.super }}
+  <a href="{% url 'security-home' %}">{% trans "Security" %}</a>
+{% endblock %}
+
+{% block appbar %}
+  <h2>{% trans 'Security' %}</h2>
+{% endblock %}
+
+{% block content %}
+  <div class="section" id="security-settings">
+    <h3>{% trans "Settings" %}</h3>
+    <div>
+      <p>{% trans "Advised default policy:" %} <tt class="default-policy">{{ view.default_policy }}</tt>
+      </p>
+      <p>
+        <button href="#" data-policy="{{ view.default_policy }}" class="set-content-security-policy-report-only">Copy in Content Security Policy Report Only</button>.
+      </p>
+      <p>
+        <form action="." method="post" id="setting-form">
+          {% csrf_token %}
+          {{ form.as_p }}
+          <div class="buttons">
+            <button>{% trans "Submit" %}</button>
+          </div>
+        </form>
+      </p>
+    </div>
+  </div>
+  {% if reports %}
+    <div class="section" id="csp-reports">
+      <h3>{% blocktrans with count=reports.count %}Security reports ({{ count }} reports){% endblocktrans %}</h3>
+      <table class="main">
+        <thead>
+          <tr>
+            <th>{% trans "Last seen" %}</th>
+            <th>{% trans "First seen" %}</th>
+            <th>{% trans "Source" %}</th>
+            <th>{% trans "Violated directive" %}</th>
+          </tr>
+        </thead>
+        <tbody>
+          {% for report in reports %}
+            <tr{% if report.error %} class="error"{% endif %} title="{{ report.full }}">
+              <td>{{ report.last_seen }}</td>
+              <td>{{ report.first_seen }}</td>
+              <td>
+                {% if report.inline %}
+                  Inline:
+                {% endif %}
+                {% with source=report.source %}
+                  <a href="{{ source }}">{{ source }}</a>
+                {% endwith %}
+                {% if report.inline %}
+                  Line: {{ report.line }} Column: {{ report.column }}
+                {% endif %}
+                {% if report.error %}<span title="error">⚠</span>{% endif %}
+              </td>
+              <td><tt>{{ report.violated_directive }}</tt></td>
+            </tr>
+          {% endfor %}
+        </tbody>
+      </table>
+      <div>
+        <form action="." method="post" id="csp-reports-form">
+          {% csrf_token %}
+          <div class="buttons">
+            <button name="clean-reports">{% trans "Clean reports" %}</button>
+          </div>
+        </form>
+      </div>
+  {% endif %}
+  </div>
+{% endblock %}

hobo/security/urls.py

@@ -0,0 +1,26 @@
+# hobo - portal to configure and deploy applications
+# Copyright (C) Entr'ouvert
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+from django.urls import path
+
+from hobo.views import admin_required
+
+from . import views
+
+urlpatterns = [
+    path('security/', admin_required(views.home), name='security-home'),
+    path('api/csp-report/', views.api_report, name='security-report'),
+]

hobo/security/utils.py

@@ -0,0 +1,35 @@
+# hobo - portal to configure and deploy applications
+# Copyright (C) 2015-2022 Entr'ouvert
+
+import urllib.parse
+
+
+def common_domains(urls):
+    '''Compute fewest common domains between urls'''
+    common = None
+    for url in urls:
+        parsed = urllib.parse.urlparse(url)
+        new = parsed.netloc.split('.')
+        if common is None:
+            common = new
+        else:
+            j = len(common)
+            for i, (a, b) in enumerate(zip(common[::-1], new[::-1])):
+                if a != b:
+                    j = i
+                    break
+            if j <= len(common):
+                common = common[::-1][:j][::-1]
+
+    new_urls = []
+    if common:
+        common_domain = '.' + '.'.join(common)
+        common_wildcard_domain = 'https://*' + common_domain
+
+    for url in urls:
+        if common and common_domain in url:
+            if common_wildcard_domain not in new_urls:
+                new_urls.append(common_wildcard_domain)
+        else:
+            new_urls.append(url)
+    return new_urls

hobo/security/views.py

@@ -0,0 +1,123 @@
+# hobo - portal to configure and deploy applications
+# Copyright (C) 2015-2022 Entr'ouvert
+
+import json
+import logging
+import re
+
+from django.contrib import messages
+from django.http import HttpResponse, HttpResponseBadRequest
+from django.urls import reverse_lazy
+from django.utils.functional import cached_property
+from django.utils.translation import gettext as _
+from django.views.decorators.csrf import csrf_exempt
+from django.views.generic import FormView
+
+from hobo.environment.utils import get_operational_services, get_setting_variable
+
+from .forms import SecurityForm
+from .models import CspReport
+from .utils import common_domains
+
+logger = logging.getLogger('hobo.security')
+
+
+REPLACE_SEMICOMMAS_BY_NEWLINES_RE = re.compile(' *; *')
+
+
+def add_newlines_to_policy(policy):
+    '''Add some spacing to policies for presentation'''
+    return re.sub(REPLACE_SEMICOMMAS_BY_NEWLINES_RE, ';\n', policy)
+
+
+class HomeView(FormView):
+    template_name = 'hobo/security/home.html'
+    form_class = SecurityForm
+    success_url = reverse_lazy('security-home')
+
+    @cached_property
+    def content_security_policy_variable(self):
+        return get_setting_variable('CONTENT_SECURITY_POLICY')
+
+    @cached_property
+    def content_security_policy_report_only_variable(self):
+        return get_setting_variable('CONTENT_SECURITY_POLICY_REPORT_ONLY')
+
+    @cached_property
+    def content_security_policy_report_uri_variable(self):
+        return get_setting_variable('CONTENT_SECURITY_POLICY_REPORT_URI')
+
+    def get_initial(self):
+        initial = super().get_initial()
+        initial['content_security_policy'] = add_newlines_to_policy(
+            self.content_security_policy_variable.value,
+        )
+        initial['content_security_policy_report_only'] = add_newlines_to_policy(
+            self.content_security_policy_report_only_variable.value,
+        )
+        return initial
+
+    def form_valid(self, form):
+        if 'clean-reports' in self.request.POST:
+            CspReport.objects.all().delete()
+            messages.info(self.request, _('All reports removed.'))
+            return super().form_valid(form)
+
+        self.content_security_policy_variable.update_value(form.cleaned_data['content_security_policy'])
+        self.content_security_policy_report_only_variable.update_value(
+            form.cleaned_data['content_security_policy_report_only']
+        )
+        self.content_security_policy_report_uri_variable.update_value(
+            self.request.build_absolute_uri('/api/csp-report/')
+        )
+        return super().form_valid(form)
+
+    def default_policy(self):
+        service_base_urls = [service.base_url for service in get_operational_services()]
+        policy = 'default-src '
+        policy += ' '.join(common_domains(service_base_urls))
+        policy += " 'unsafe-eval'"
+        policy += " 'unsafe-inline'"
+        return policy
+
+    def get_context_data(self, **kwargs):
+        ctx = super().get_context_data(**kwargs)
+        ctx['reports'] = CspReport.objects.order_by('-last_seen').only(
+            'first_seen', 'last_seen', 'source', 'violated_directive'
+        )
+        return ctx
+
+
+home = HomeView.as_view()
+
+
+@csrf_exempt
+def api_report(request):
+    content_security_policy_variable = get_setting_variable('CONTENT_SECURITY_POLICY')
+    content_security_policy_report_only_variable = get_setting_variable('CONTENT_SECURITY_POLICY_REPORT_ONLY')
+
+    # do not accept reports if no policy is published
+    if not content_security_policy_variable.value and not content_security_policy_report_only_variable.value:
+        return HttpResponseBadRequest('no CSP')
+
+    if CspReport.too_much_reports():
+        return HttpResponseBadRequest('too much reports')
+
+    # gather values for the report
+    try:
+        error = 'error' in request.GET
+        report = json.loads(request.body)
+        csp_report = CspReport.record_from_report(report, error=error)
+    except Exception as e:
+        logger.warning('invalid CSP report: %s, %r', e, request.body[:1024])
+        return HttpResponseBadRequest('invalid json')
+
+    # delete reports last_seen more than one day ago
+    CspReport.clean_old_reports(days=1)
+
+    # log an error if ?error is present, ?error is added to the
+    # report-uri by the middleware for blocking CSP policies
+    if error and content_security_policy_variable:
+        hostname = request.headers['host']
+        logger.error(f'{hostname}: CSP reports %s', csp_report)
+    return HttpResponse()

hobo/settings.py

@@ -52,6 +52,7 @@ INSTALLED_APPS = (
     'hobo.maintenance',
     'hobo.matomo',
     'hobo.profile',
+    'hobo.security',
     'hobo.seo',
     'hobo.theme',
     'hobo.emails',

hobo/static/css/style.scss

@@ -449,3 +449,12 @@ div.app-parameters--values label {
 #theme-filter {
 	margin-bottom: 1em;
 }
+#security-settings{
+    textarea {
+        width: 100%;
+    }
+    .default-policy {
+        background: lightgray;
+        padding: 0.3ex 0.3ex;
+    }
+}

hobo/static/js/hobo.js

@@ -55,4 +55,10 @@ $(function() {
       $('#themes-list > div').show();
     }
   });
+
+  $('body').delegate('#security-settings .set-content-security-policy-report-only', 'click', function (ev) {
+      var $target = $(ev.target);
+      var policy = $target.data('policy');
+      $('#security-settings [name="content_security_policy_report_only"]').val(policy);
+  })
 });

hobo/templates/hobo/home.html

@@ -117,5 +117,6 @@
     <a class="button button-paragraph" href="{% url 'environment-variables' %}">{% trans 'Variables' %}</a>
     <a class="button button-paragraph" href="{% url 'debug-home' %}">{% trans 'Debugging' %}</a>
     {% if show_maintenance_menu %}<a class="button button-paragraph" href="{% url 'maintenance-home' %}">{% trans 'Maintenance' %}</a>{% endif %}
+    <a class="button button-paragraph" href="{% url 'security-home' %}">{% trans 'Security' %}</a>
   </aside>
 {% endblock %}

hobo/urls.py

@@ -24,6 +24,7 @@ from .environment.urls import urlpatterns as environment_urls
 from .maintenance.urls import urlpatterns as maintenance_urls
 from .matomo.urls import urlpatterns as matomo_urls
 from .profile.urls import urlpatterns as profile_urls
+from .security.urls import urlpatterns as security_urls
 from .seo.urls import urlpatterns as seo_urls
 from .sms.urls import urlpatterns as sms_urls
 from .theme.urls import urlpatterns as theme_urls
@@ -42,6 +43,7 @@ urlpatterns = [
     re_path(r'^debug/', decorated_includes(admin_required, include(debug_urls))),
     re_path(r'^applications/', decorated_includes(admin_required, include(applications_urls))),
     re_path(r'^maintenance/', decorated_includes(admin_required, include(maintenance_urls))),
+    path('', include(security_urls)),
     path('api/health/', health_json, name='health-json'),
     re_path(r'^menu.json$', menu_json, name='menu_json'),
     re_path(r'^hobos.json$', hobo),

tests/settings.py

@@ -22,6 +22,7 @@ MIDDLEWARE = MIDDLEWARE + (
     'hobo.middleware.RobotsTxtMiddleware',
     'hobo.provisionning.middleware.ProvisionningMiddleware',
     'hobo.middleware.maintenance.MaintenanceMiddleware',
+    'hobo.middleware.security.content_security_policy_middleware',
 )
 
 common_middleware_index = MIDDLEWARE.index('django.middleware.common.CommonMiddleware')

tests/test_security.py

@@ -0,0 +1,167 @@
+# hobo - portal to configure and deploy applications
+# Copyright (C) Entr'ouvert
+
+import unittest.mock
+
+from django.core.cache import cache
+
+from hobo.environment.models import Variable
+from hobo.environment.utils import get_setting_variable
+from hobo.security.models import CspReport
+
+from .test_manager import login
+
+CSP_HEADER_NAME = 'Content-Security-Policy'
+CSP_REPORT_ONLY_HEADER_NAME = 'Content-Security-Policy-Report-Only'
+
+POLICY = "default-src *.example.com 'unsafe-eval' 'unsafe-inline'"
+POLICY2 = "default-src https://*.example.com 'unsafe-eval' 'unsafe-inline'"
+
+# Copied from https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Security-Policy-Report-Only
+CSP_REPORT = {
+    'csp-report': {
+        'blocked-uri': 'http://example.com/css/style.css',
+        'disposition': 'report',
+        'document-uri': 'http://example.com/signup.html',
+        'effective-directive': 'style-src-elem',
+        'original-policy': "default-src 'none'; style-src cdn.example.com; report-uri /_/csp-reports",
+        'referrer': '',
+        'status-code': 200,
+        'violated-directive': 'style-src-elem',
+    }
+}
+
+CSP_REPORT_INLINE = {
+    'csp-report': {
+        'blocked-uri': 'inline',
+        'disposition': 'report',
+        'document-uri': 'http://example.com/signup.html',
+        'effective-directive': 'style-src-elem',
+        'original-policy': "default-src 'none'; style-src cdn.example.com; report-uri /_/csp-reports",
+        'referrer': '',
+        'status-code': 200,
+        'violated-directive': 'style-src-elem',
+    }
+}
+
+
+def test_middleware(app, settings):
+    resp = app.get('/')
+    assert CSP_HEADER_NAME not in resp.headers
+    assert CSP_REPORT_ONLY_HEADER_NAME not in resp.headers
+
+    settings.CONTENT_SECURITY_POLICY = POLICY
+    resp = app.get('/')
+    assert CSP_HEADER_NAME in resp.headers
+    assert CSP_REPORT_ONLY_HEADER_NAME not in resp.headers
+    assert resp.headers[CSP_HEADER_NAME] == POLICY
+
+    settings.CONTENT_SECURITY_POLICY_REPORT_ONLY = POLICY
+    resp = app.get('/')
+    assert CSP_HEADER_NAME in resp.headers
+    assert CSP_REPORT_ONLY_HEADER_NAME in resp.headers
+    assert resp.headers[CSP_HEADER_NAME] == POLICY
+    assert resp.headers[CSP_REPORT_ONLY_HEADER_NAME] == POLICY
+
+    settings.CONTENT_SECURITY_POLICY_REPORT_URI = 'https://csp.example.com/csp-report/'
+    resp = app.get('/')
+    assert CSP_HEADER_NAME in resp.headers
+    assert CSP_REPORT_ONLY_HEADER_NAME in resp.headers
+    assert resp.headers[CSP_HEADER_NAME] == POLICY + '; report-uri https://csp.example.com/csp-report/?error'
+    assert (
+        resp.headers[CSP_REPORT_ONLY_HEADER_NAME]
+        == POLICY + '; report-uri https://csp.example.com/csp-report/'
+    )
+
+
+def test_manage(app, admin_user, settings):
+    assert Variable.objects.filter(name='SETTING_CONTENT_SECURITY_POLICY').count() == 0
+    assert Variable.objects.filter(name='SETTING_CONTENT_SECURITY_POLICY_REPORT_ONLY').count() == 0
+
+    login(app)
+    resp = app.get('/')
+    resp = resp.click('Security')
+
+    form = resp.forms['setting-form']
+    form.set('content_security_policy', POLICY)
+    form.set('content_security_policy_report_only', POLICY2)
+    resp = form.submit()
+
+    assert 'errorlist' not in resp
+    resp = resp.follow()
+
+    assert Variable.objects.filter(name='SETTING_CONTENT_SECURITY_POLICY').get().value == POLICY
+    assert Variable.objects.filter(name='SETTING_CONTENT_SECURITY_POLICY_REPORT_ONLY').get().value == POLICY2
+    assert Variable.objects.filter(name='SETTING_CONTENT_SECURITY_POLICY_REPORT_URI').get().value == (
+        'http://testserver/api/csp-report/'
+    )
+
+    form = resp.forms['setting-form']
+    form.set('content_security_policy', '')
+    form.set('content_security_policy_report_only', '')
+    resp = form.submit()
+
+    assert Variable.objects.filter(name='SETTING_CONTENT_SECURITY_POLICY').get().value == ''
+    assert Variable.objects.filter(name='SETTING_CONTENT_SECURITY_POLICY_REPORT_ONLY').get().value == ''
+
+
+def test_report_table(app, admin_user, settings, freezer):
+    freezer.move_to('2024-01-17 10:00:00+01:00')
+
+    CspReport.record_from_report(CSP_REPORT)
+    CspReport.record_from_report(CSP_REPORT_INLINE, error=True)
+
+    login(app)
+    resp = app.get('/security/')
+    rows = resp.pyquery.items('tr')
+    assert [[cell.text() for cell in row.items('td,th')] for row in rows] == [
+        ['Last seen', 'First seen', 'Source', 'Violated directive'],
+        [
+            'Jan. 17, 2024, 9 a.m.',
+            'Jan. 17, 2024, 9 a.m.',
+            'http://example.com/css/style.css',
+            'style-src-elem',
+        ],
+        [
+            'Jan. 17, 2024, 9 a.m.',
+            'Jan. 17, 2024, 9 a.m.',
+            'Inline: http://example.com/signup.html Line: 0 Column: 0 ⚠',
+            'style-src-elem',
+        ],
+    ]
+
+    resp = resp.forms['csp-reports-form'].submit(name='clean-reports').follow()
+    rows = resp.pyquery.items('tr')
+    assert [[cell.text() for cell in row.items('td,th')] for row in rows][1:] == []
+
+
+def test_api_report(app, db, caplog, freezer):
+    caplog.set_level('WARNING')
+    app.post_json('/api/csp-report/', params=CSP_REPORT, status=400)
+
+    content_security_policy_variable = get_setting_variable('CONTENT_SECURITY_POLICY')
+    content_security_policy_variable.value = POLICY
+    content_security_policy_variable.save()
+
+    app.post('/api/csp-report/', params=b'', status=400)
+    caplog.clear()
+
+    app.post_json('/api/csp-report/', params=CSP_REPORT, status=200)
+    assert CspReport.objects.get().content == CSP_REPORT['csp-report']
+    assert not caplog.records
+    caplog.set_level('ERROR')
+    app.post_json('/api/csp-report/?error', params=CSP_REPORT, status=200)
+    assert caplog.messages == [
+        'testserver: CSP reports blocked asset http://example.com/css/style.css by directive style-src-elem'
+    ]
+
+    cache.clear()
+
+    # set too-much-reports condition
+    with unittest.mock.patch('hobo.security.models.CspReport.objects.count', lambda *args: 1001):
+        app.post_json('/api/csp-report/', params=CSP_REPORT, status=400)
+    # too-much-reports is cached
+    app.post_json('/api/csp-report/', params=CSP_REPORT, status=400)
+    freezer.tick(61)
+    # too-much-reports cache is expired
+    app.post_json('/api/csp-report/', params=CSP_REPORT, status=200)