diff --git a/app/access.py b/app/access.py new file mode 100644 index 0000000..c89a255 --- /dev/null +++ b/app/access.py @@ -0,0 +1,61 @@ +import secrets + + +GUEST_SCAN_CAPABILITIES_SESSION_KEY = 'guest_scan_capabilities' +GUEST_APP_CAPABILITIES_SESSION_KEY = 'guest_app_capabilities' + + +def grant_guest_scan_access(request, scan): + """Grant this browser an unguessable, session-bound capability for a scan.""" + capabilities = request.session.get(GUEST_SCAN_CAPABILITIES_SESSION_KEY, {}) + capabilities[str(scan.pk)] = secrets.token_urlsafe(32) + request.session[GUEST_SCAN_CAPABILITIES_SESSION_KEY] = capabilities + request.session.modified = True + + +def grant_guest_app_access(request, app): + """Grant this browser an unguessable, session-bound capability for an app.""" + capabilities = request.session.get(GUEST_APP_CAPABILITIES_SESSION_KEY, {}) + capabilities[str(app.pk)] = secrets.token_urlsafe(32) + request.session[GUEST_APP_CAPABILITIES_SESSION_KEY] = capabilities + request.session.modified = True + + +def guest_scan_ids(request): + capabilities = request.session.get(GUEST_SCAN_CAPABILITIES_SESSION_KEY, {}) + if not isinstance(capabilities, dict): + return [] + + return [int(scan_id) for scan_id in capabilities if scan_id.isdigit()] + + +def guest_app_ids(request): + capabilities = request.session.get(GUEST_APP_CAPABILITIES_SESSION_KEY, {}) + if not isinstance(capabilities, dict): + return [] + + return [int(app_id) for app_id in capabilities if app_id.isdigit()] + + +def can_access_app(request, app): + if app.user_id is not None: + return request.user.is_authenticated and app.user_id == request.user.id + + capabilities = request.session.get(GUEST_APP_CAPABILITIES_SESSION_KEY, {}) + if not isinstance(capabilities, dict): + return False + + capability = capabilities.get(str(app.pk)) + return isinstance(capability, str) and bool(capability) + + +def can_access_scan(request, scan): + if scan.user_id is not None: + return request.user.is_authenticated and scan.user_id == request.user.id + + capabilities = request.session.get(GUEST_SCAN_CAPABILITIES_SESSION_KEY, {}) + if not isinstance(capabilities, dict): + return False + + capability = capabilities.get(str(scan.pk)) + return isinstance(capability, str) and bool(capability) diff --git a/app/api.py b/app/api.py index 5da9b82..89944f6 100644 --- a/app/api.py +++ b/app/api.py @@ -7,30 +7,73 @@ from app.worker.tasks import task_create_scan from rest_framework import viewsets, mixins, status from django.db.models import Q +from django.shortcuts import get_object_or_404 from django_filters import rest_framework as filters +from app.access import ( + can_access_app, + grant_guest_app_access, + grant_guest_scan_access, + guest_app_ids, + guest_scan_ids, +) +from rest_framework.exceptions import PermissionDenied + + +class IsAuthenticatedOrGuestCreate(permissions.BasePermission): + """Allow public creation while leaving unsafe object changes owner-only.""" + + def has_permission(self, request, view): + return request.method in permissions.SAFE_METHODS or request.method == 'POST' or request.user.is_authenticated class IsUserOrReadOnly(permissions.BasePermission): def has_object_permission(self, request, view, obj): - if request.method in ['PUT', 'PATCH']: - return obj.user == request.user - return True + if request.method in permissions.SAFE_METHODS: + return True + return request.user.is_authenticated and obj.user == request.user class ApplicationViewSet(viewsets.ModelViewSet): serializer_class = ApplicationSerializer queryset = Application.objects.all() - permission_classes = (permissions.IsAuthenticatedOrReadOnly, IsUserOrReadOnly) + permission_classes = (IsAuthenticatedOrGuestCreate, IsUserOrReadOnly) + + def get_queryset(self): + if self.request.user.is_authenticated: + return Application.objects.filter( + Q(user=self.request.user) | Q(pk__in=guest_app_ids(self.request)) + ) + return Application.objects.filter(pk__in=guest_app_ids(self.request)) def perform_create(self, serializer): - obj = serializer.save(user=self.request.user) + if self.request.user.is_authenticated: + serializer.save(user=self.request.user) + else: + app = serializer.save() + grant_guest_app_access(self.request, app) class ScanViewSet(viewsets.ModelViewSet): serializer_class = ScanSerializer queryset = Scan.objects.all() - permission_classes = (permissions.IsAuthenticatedOrReadOnly, IsUserOrReadOnly) + permission_classes = (IsAuthenticatedOrGuestCreate, IsUserOrReadOnly) parser_classes = (MultiPartParser, FormParser) + + def get_queryset(self): + if self.request.user.is_authenticated: + return Scan.objects.filter( + Q(user=self.request.user) | Q(pk__in=guest_scan_ids(self.request)) + ) + return Scan.objects.filter(pk__in=guest_scan_ids(self.request)) def perform_create(self, serializer): - scan = serializer.save(user=self.request.user, status='In progress', progress=1) + app = serializer.validated_data.get('app') + if app is not None and not can_access_app(self.request, app): + raise PermissionDenied('You do not have access to this application.') + + save_kwargs = {'status': 'In progress', 'progress': 1} + if self.request.user.is_authenticated and (app is None or app.user_id is not None): + save_kwargs['user'] = self.request.user + scan = serializer.save(**save_kwargs) + if scan.user_id is None: + grant_guest_scan_access(self.request, scan) task_id = task_create_scan.delay(scan.id) scan.task = task_id.id scan.save() @@ -40,6 +83,13 @@ class FindingViewSet(viewsets.ModelViewSet): queryset = Finding.objects.all() permission_classes = (permissions.IsAuthenticatedOrReadOnly, IsUserOrReadOnly) + def get_queryset(self): + if self.request.user.is_authenticated: + return Finding.objects.filter( + Q(scan__user=self.request.user) | Q(scan_id__in=guest_scan_ids(self.request)) + ) + return Finding.objects.filter(scan_id__in=guest_scan_ids(self.request)) + def perform_create(self, serializer): obj = serializer.save(user=self.request.user) @@ -47,7 +97,12 @@ def perform_create(self, serializer): @action(detail=True, methods=['GET'], name='Get findings for scan') def scan(self, request, pk=None): if (pk != None): - scan = Scan.objects.get(pk=pk) + scan_queryset = Scan.objects.filter(pk__in=guest_scan_ids(request)) + if request.user.is_authenticated: + scan_queryset = Scan.objects.filter( + Q(user=request.user) | Q(pk__in=guest_scan_ids(request)) + ) + scan = get_object_or_404(scan_queryset, pk=pk) queryset = Finding.objects.filter(scan=scan).order_by('id') else: queryset = Finding.objects.all().order_by('id') @@ -64,6 +119,13 @@ class PermissionViewSet(viewsets.ModelViewSet): queryset = Permission.objects.all() permission_classes = (permissions.IsAuthenticatedOrReadOnly, IsUserOrReadOnly) + def get_queryset(self): + if self.request.user.is_authenticated: + return Permission.objects.filter( + Q(scan__user=self.request.user) | Q(scan_id__in=guest_scan_ids(self.request)) + ) + return Permission.objects.filter(scan_id__in=guest_scan_ids(self.request)) + def perform_create(self, serializer): obj = serializer.save(user=self.request.user) @@ -71,7 +133,12 @@ def perform_create(self, serializer): @action(detail=True, methods=['GET'], name='Get findings for scan') def scan(self, request, pk=None): if (pk != None): - scan = Scan.objects.get(pk=pk) + scan_queryset = Scan.objects.filter(pk__in=guest_scan_ids(request)) + if request.user.is_authenticated: + scan_queryset = Scan.objects.filter( + Q(user=request.user) | Q(pk__in=guest_scan_ids(request)) + ) + scan = get_object_or_404(scan_queryset, pk=pk) queryset = Permission.objects.filter(scan=scan).order_by('id') else: queryset = Permission.objects.all().order_by('id') diff --git a/app/config/urls.py b/app/config/urls.py index 1e46002..c29c8db 100755 --- a/app/config/urls.py +++ b/app/config/urls.py @@ -5,7 +5,6 @@ from rest_framework.authtoken.views import obtain_auth_token from drf_yasg.views import get_schema_view from drf_yasg import openapi -from app.worker.tasks import scan_state schema_view = get_schema_view( openapi.Info( @@ -55,9 +54,9 @@ path('accounts/profile/', views.user_profile, name='profile'), path('api/v1/auth-token/', obtain_auth_token, name='api_token_auth'), path('api/v1/', include(router.urls)), - path('scan_state/', scan_state, name="scan_state"), + path('scan_state/', views.scan_state, name="scan_state"), re_path(r'^swagger(?P\.json|\.yaml)$', schema_view.without_ui(cache_timeout=0), name='schema-json'), re_path(r'^swagger/$', schema_view.with_ui('swagger', cache_timeout=0), name='schema-swagger-ui'), re_path(r'^redoc/$', schema_view.with_ui('redoc', cache_timeout=0), name='schema-redoc'), -] \ No newline at end of file +] diff --git a/app/forms.py b/app/forms.py index 5a35faa..d621a98 100755 --- a/app/forms.py +++ b/app/forms.py @@ -43,6 +43,12 @@ class ScanForm(forms.ModelForm): class Meta: model = Scan fields = ('description', 'apk', 'app', 'defectdojo_id') + + +class GuestScanForm(forms.ModelForm): + class Meta: + model = Scan + fields = ('description', 'apk') class ApplicationForm(forms.ModelForm): class Meta: @@ -55,4 +61,4 @@ class FindingForm(forms.ModelForm): risk = RiskModelChoiceField(queryset=Risk.objects.all()) class Meta: model = Finding - fields = ('scan', 'name', 'description', 'severity', 'status', 'path', 'line_number', 'line', 'snippet', 'cwe', 'risk', 'mitigation', 'defectdojo_id') \ No newline at end of file + fields = ('scan', 'name', 'description', 'severity', 'status', 'path', 'line_number', 'line', 'snippet', 'cwe', 'risk', 'mitigation', 'defectdojo_id') diff --git a/app/migrations/0004_allow_guest_scans.py b/app/migrations/0004_allow_guest_scans.py new file mode 100644 index 0000000..cd64ded --- /dev/null +++ b/app/migrations/0004_allow_guest_scans.py @@ -0,0 +1,34 @@ +# Generated by Django 3.2 on 2026-07-28 + +from django.db import migrations, models +import django.db.models.deletion + + +class Migration(migrations.Migration): + + dependencies = [ + ('app', '0003_auto_20210509_1408'), + ] + + operations = [ + migrations.AlterField( + model_name='scan', + name='app', + field=models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.CASCADE, + to='app.application', + ), + ), + migrations.AlterField( + model_name='scan', + name='user', + field=models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.PROTECT, + to='auth.user', + ), + ), + ] diff --git a/app/migrations/0005_allow_guest_apps.py b/app/migrations/0005_allow_guest_apps.py new file mode 100644 index 0000000..171ee92 --- /dev/null +++ b/app/migrations/0005_allow_guest_apps.py @@ -0,0 +1,24 @@ +# Generated by Django 3.2 on 2026-08-26 + +from django.db import migrations, models +import django.db.models.deletion + + +class Migration(migrations.Migration): + + dependencies = [ + ('app', '0004_allow_guest_scans'), + ] + + operations = [ + migrations.AlterField( + model_name='application', + name='user', + field=models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.PROTECT, + to='auth.user', + ), + ), + ] diff --git a/app/models.py b/app/models.py index 7a58e12..3ea06f4 100755 --- a/app/models.py +++ b/app/models.py @@ -52,18 +52,18 @@ class Application(models.Model): id = models.AutoField(primary_key=True) name = models.CharField(max_length=255, blank=False) description = models.CharField(max_length=255, blank=False) - user = models.ForeignKey(User, on_delete=models.PROTECT) + user = models.ForeignKey(User, on_delete=models.PROTECT, blank=True, null=True) created_on = models.DateTimeField(auto_now_add=True, null=True) updated_on = models.DateTimeField(auto_now=True, null=True) class Scan(models.Model): id = models.AutoField(primary_key=True) - app = models.ForeignKey(Application, on_delete=models.CASCADE) + app = models.ForeignKey(Application, on_delete=models.CASCADE, blank=True, null=True) name = models.CharField(max_length=255, blank=True) apk = models.FileField(upload_to='apk/', blank=False, validators=[validate_file_extension]) description = models.CharField(max_length=255, blank=False) defectdojo_id = models.IntegerField(blank=True, default=0) - user = models.ForeignKey(User, on_delete=models.PROTECT) + user = models.ForeignKey(User, on_delete=models.PROTECT, blank=True, null=True) status = models.CharField(max_length=50,blank=True, null=True) progress = models.IntegerField(blank=True, null=True) findings = models.IntegerField(blank=True, null=True, default=0) @@ -293,4 +293,4 @@ class Antivirus(models.Model): method = models.CharField(max_length=255, null=True) update = models.CharField(max_length=255, null=True) created_on = models.DateTimeField(auto_now_add=True, null=True) - updated_on = models.DateTimeField(auto_now=True, null=True) \ No newline at end of file + updated_on = models.DateTimeField(auto_now=True, null=True) diff --git a/app/templates/app.html b/app/templates/app.html index 857fce1..59b2ba0 100755 --- a/app/templates/app.html +++ b/app/templates/app.html @@ -14,7 +14,7 @@

App

Created by - {{ app.user }} + {% if app.user %}{{ app.user }}{% else %}Guest{% endif %} diff --git a/app/templates/create_app.html b/app/templates/create_app.html index c9586ce..3e5e74e 100755 --- a/app/templates/create_app.html +++ b/app/templates/create_app.html @@ -37,4 +37,4 @@ }); {% endif %} -{% endblock %} \ No newline at end of file +{% endblock %} diff --git a/app/templates/create_scan.html b/app/templates/create_scan.html index a62d1d6..c7e3779 100755 --- a/app/templates/create_scan.html +++ b/app/templates/create_scan.html @@ -37,4 +37,4 @@ fade_alerts(); }); -{% endblock %} \ No newline at end of file +{% endblock %} diff --git a/app/templates/home.html b/app/templates/home.html index 5fee322..7170044 100755 --- a/app/templates/home.html +++ b/app/templates/home.html @@ -4,8 +4,8 @@
Apps
New App {% else %} New App {% endif %} {% else %} - -New App {% endif %} +New App +{% endif %}
{% if apps %}
    {% for message in messages %} @@ -31,7 +31,7 @@
    Apps
    {{ app.id }} {{ app.name }} - {{ app.user.username }} + {% if app.user %}{{ app.user.username }}{% else %}Guest{% endif %} {{ app.description }} @@ -101,7 +101,7 @@
    Apps
    {% endif %} {% endfor %}
    - {% if user.is_authenticated %} + {% if user.is_authenticated or not app.user %} New Scan {% endif %} @@ -131,4 +131,4 @@
    Apps
    }); }); -{% endif %} {% endblock %} \ No newline at end of file +{% endif %} {% endblock %} diff --git a/app/templates/scan.html b/app/templates/scan.html index bb89e16..1d6b15b 100644 --- a/app/templates/scan.html +++ b/app/templates/scan.html @@ -51,9 +51,11 @@

    Mobile Audit

  • {% fa5_icon 'database' %} Databases
  • + {% if not guest_scan %}
  • {% fa5_icon 'download' %} Export
  • + {% endif %}
@@ -69,7 +71,7 @@

Mobile Audit

Scan

Description: {{ scan.description }}

-

Created by: {{ scan.user.username }}

+

Created by: {% if scan.user %}{{ scan.user.username }}{% else %}Guest{% endif %}

Status: {{ scan.status }}

{% if scan.progress == 100 %} @@ -80,6 +82,7 @@

Scan


+ {% if not guest_scan %}
{% csrf_token %} @@ -87,6 +90,7 @@

Scan

{% if scan.progress >= 40 %} Export {% fa5_icon 'download' %} {% endif %} + {% endif %}
@@ -960,4 +964,4 @@
Databases
}); }); -{% endblock %} \ No newline at end of file +{% endblock %} diff --git a/app/tests.py b/app/tests.py index 37a537c..eaa40d3 100755 --- a/app/tests.py +++ b/app/tests.py @@ -6,9 +6,14 @@ import tempfile from unittest.mock import MagicMock, call, patch -from django.test import SimpleTestCase, override_settings +from django.contrib.auth.models import User +from django.core.files.uploadedfile import SimpleUploadedFile +from django.test import SimpleTestCase, TestCase, override_settings +from django.urls import reverse +from rest_framework.test import APIClient from app import analysis +from app.models import Application, Scan from app.worker import tasks @@ -275,3 +280,175 @@ def test_scan_state_returns_ordinary_result_dict_unchanged( response = tasks.scan_state(MagicMock(), 1) self.assertEqual(json.loads(response.content), progress) + + +class GuestScanAccessTests(TestCase): + @patch('app.views.task_create_scan.delay') + def test_guest_can_start_a_scan_and_only_its_session_can_access_it(self, delay): + delay.return_value.id = 'guest-scan-task' + apk = SimpleUploadedFile( + 'guest.apk', + b'APK test data', + content_type='application/vnd.android.package-archive', + ) + + response = self.client.post( + reverse('create_scan'), + {'description': 'Guest upload', 'apk': apk}, + ) + + self.assertEqual(response.status_code, 302) + scan = Scan.objects.get(description='Guest upload') + self.assertIsNone(scan.user) + self.assertIsNone(scan.app) + self.assertEqual(response['Location'], reverse('scan', kwargs={'id': scan.id})) + + self.assertEqual(self.client.get(reverse('scan', kwargs={'id': scan.id})).status_code, 200) + self.assertEqual(self.client.get(reverse('scan_state', kwargs={'id': scan.id})).status_code, 200) + self.assertEqual(self.client.get('/api/v1/scan/{}/'.format(scan.id)).status_code, 200) + + other_browser = self.client_class() + self.assertEqual(other_browser.get(reverse('scan', kwargs={'id': scan.id})).status_code, 404) + self.assertEqual(other_browser.get(reverse('scan_state', kwargs={'id': scan.id})).status_code, 404) + self.assertEqual(other_browser.get('/api/v1/scan/{}/'.format(scan.id)).status_code, 404) + + def test_other_authenticated_browser_cannot_export_guest_scan(self): + scan = Scan.objects.create( + description='Guest upload', + apk=SimpleUploadedFile('guest.apk', b'APK test data'), + ) + other_browser = self.client_class() + other_browser.force_login(User.objects.create_user('other', password='password')) + + response = other_browser.get(reverse('export', kwargs={'id': scan.id})) + + self.assertEqual(response.status_code, 404) + + @patch('app.views.get_report_virus_total') + def test_virustotal_update_requires_authenticated_authorized_access(self, get_report): + scan = Scan.objects.create( + description='Guest upload', + apk=SimpleUploadedFile('guest.apk', b'APK test data'), + ) + url = reverse('update_virustotal', kwargs={'scan_id': scan.id}) + + self.assertEqual(self.client.get(url).status_code, 302) + self.assertFalse(get_report.called) + + other_browser = self.client_class() + other_browser.force_login(User.objects.create_user('other', password='password')) + self.assertEqual(other_browser.get(url).status_code, 404) + self.assertFalse(get_report.called) + + @patch('app.views.get_report_virus_total') + def test_authenticated_user_can_update_own_scan_virustotal_report(self, get_report): + user = User.objects.create_user('owner', password='password') + scan = Scan.objects.create( + user=user, + description='Signed-in upload', + apk=SimpleUploadedFile('signed-in.apk', b'APK test data'), + sha256='sha256', + ) + self.client.force_login(user) + + response = self.client.get( + reverse('update_virustotal', kwargs={'scan_id': scan.id}) + ) + + self.assertRedirects(response, reverse('scan', kwargs={'id': scan.id})) + get_report.assert_called_once_with(scan, 'sha256') + + +class GuestApplicationAccessTests(TestCase): + @patch('app.views.task_create_scan.delay') + def test_guest_app_and_its_scan_are_private_to_the_creating_session(self, delay): + delay.return_value.id = 'guest-app-scan-task' + + response = self.client.post( + reverse('create_app'), + {'name': 'Guest application', 'description': 'Created without an account'}, + ) + + self.assertEqual(response.status_code, 302) + app = Application.objects.get(name='Guest application') + self.assertIsNone(app.user) + self.assertEqual( + response['Location'], + reverse('create_scan', kwargs={'app_id': app.id}), + ) + self.assertEqual(self.client.get(reverse('app', kwargs={'id': app.id})).status_code, 200) + + apk = SimpleUploadedFile('guest-app.apk', b'APK test data') + response = self.client.post( + reverse('create_scan', kwargs={'app_id': app.id}), + {'description': 'Guest app upload', 'apk': apk}, + ) + + scan = Scan.objects.get(description='Guest app upload') + self.assertEqual(scan.app, app) + self.assertIsNone(scan.user) + self.assertEqual(self.client.get(reverse('scan', kwargs={'id': scan.id})).status_code, 200) + + other_browser = self.client_class() + self.assertEqual(other_browser.get(reverse('app', kwargs={'id': app.id})).status_code, 404) + self.assertEqual( + other_browser.get(reverse('create_scan', kwargs={'app_id': app.id})).status_code, + 404, + ) + self.assertEqual(other_browser.get(reverse('scan', kwargs={'id': scan.id})).status_code, 404) + + other_browser.force_login(User.objects.create_user('other-user', password='password')) + self.assertEqual(other_browser.get(reverse('app', kwargs={'id': app.id})).status_code, 404) + + def test_guest_can_create_an_app_through_the_api_only_for_its_session(self): + response = self.client.post( + '/api/v1/app/', + data=json.dumps({'name': 'Guest API application', 'description': 'API-created'}), + content_type='application/json', + ) + + self.assertEqual(response.status_code, 201) + app = Application.objects.get(name='Guest API application') + self.assertIsNone(app.user) + self.assertEqual(self.client.get('/api/v1/app/{}/'.format(app.id)).status_code, 200) + + other_browser = self.client_class() + self.assertEqual(other_browser.get('/api/v1/app/{}/'.format(app.id)).status_code, 404) + + +class OwnershipIsolationTests(TestCase): + def setUp(self): + self.owner = User.objects.create_user('owner', password='password') + self.other_user = User.objects.create_user('other', password='password') + self.app = Application.objects.create( + name='Owner application', + description='Private application', + user=self.owner, + ) + self.scan = Scan.objects.create( + app=self.app, + user=self.owner, + description='Private scan', + apk=SimpleUploadedFile('private.apk', b'APK test data'), + ) + + def test_authenticated_user_cannot_view_another_users_app_or_scan(self): + other_browser = self.client_class() + other_browser.force_login(self.other_user) + + self.assertEqual(other_browser.get(reverse('app', kwargs={'id': self.app.id})).status_code, 404) + self.assertEqual(other_browser.get(reverse('scan', kwargs={'id': self.scan.id})).status_code, 404) + + def test_api_hides_other_users_apps_and_scans_from_list_and_detail(self): + api_client = APIClient() + api_client.force_authenticate(user=self.other_user) + + app_list = api_client.get('/api/v1/app/') + scan_list = api_client.get('/api/v1/scan/') + + self.assertEqual(app_list.status_code, 200) + self.assertEqual(scan_list.status_code, 200) + self.assertNotIn(self.app.id, [item['id'] for item in app_list.data['results']]) + self.assertNotIn(self.scan.id, [item['id'] for item in scan_list.data['results']]) + self.assertEqual(api_client.get('/api/v1/app/{}/'.format(self.app.id)).status_code, 404) + self.assertEqual(api_client.get('/api/v1/scan/{}/'.format(self.scan.id)).status_code, 404) diff --git a/app/views.py b/app/views.py index 64f3fa2..12e9810 100755 --- a/app/views.py +++ b/app/views.py @@ -1,4 +1,4 @@ -from django.shortcuts import render, redirect +from django.shortcuts import get_object_or_404, render, redirect from django.core.files.storage import FileSystemStorage from django.urls import reverse from django.contrib import messages @@ -7,13 +7,21 @@ from django.contrib.auth.forms import AuthenticationForm from django.contrib.auth.decorators import login_required from django.conf import settings -from django.http import HttpResponse +from django.http import Http404, HttpResponse from django.template.loader import get_template import pdfkit, requests, logging -from app.forms import ScanForm, ApplicationForm, FindingForm, SignUpForm, ProfileForm +from app.access import ( + can_access_app, + can_access_scan, + grant_guest_app_access, + grant_guest_scan_access, + guest_app_ids, + guest_scan_ids, +) +from app.forms import GuestScanForm, ScanForm, ApplicationForm, FindingForm, SignUpForm, ProfileForm from app import analysis from app.models import * -from app.worker.tasks import task_create_scan +from app.worker.tasks import scan_state as get_scan_state, task_create_scan from app.integration import get_report_virus_total logger = logging.getLogger('app') @@ -77,8 +85,12 @@ def user_profile(request): }) def home(request): - apps = Application.objects.all().order_by('id') - scans = Scan.objects.all().order_by('id') + if request.user.is_authenticated: + apps = Application.objects.filter(user=request.user).order_by('id') + scans = Scan.objects.filter(user=request.user).order_by('id') + else: + apps = Application.objects.filter(pk__in=guest_app_ids(request)).order_by('id') + scans = Scan.objects.filter(pk__in=guest_scan_ids(request)).order_by('id') scans_data = {} for scan in scans: scans_data[scan.id] = { @@ -128,9 +140,10 @@ def get_components_intents(scan_id): components_intents.append((component, intents)) return components_intents -@login_required def scan(request, id): - scan = Scan.objects.get(pk=id) + scan = get_object_or_404(Scan, pk=id) + if not can_access_scan(request, scan): + raise Http404 certificates = Certificate.objects.filter(scan=id).order_by('id') permissions = Permission.objects.filter(scan=id).order_by('id') activities = Activity.objects.filter(scan=id).order_by('id') @@ -151,6 +164,7 @@ def scan(request, id): antivirus = False return render(request, 'scan.html', { 'scan' : scan, + 'guest_scan': scan.user_id is None, 'permissions': permissions, 'findings': findings, 'certificates': certificates, @@ -169,32 +183,52 @@ def scan(request, id): 'settings': settings, }) -@login_required def create_scan(request, app_id = ''): + app = None + if app_id: + app = get_object_or_404(Application, pk=app_id) + if not can_access_app(request, app): + raise Http404 if request.method == 'POST': - form = ScanForm(request.POST, request.FILES) + form_class = ScanForm if request.user.is_authenticated else GuestScanForm + form = form_class(request.POST, request.FILES) if form.is_valid(): scan = form.save(commit=False) - scan.user = request.user + # Scans attached to a guest app stay session-owned even if the + # browser signs in after creating that app. + if request.user.is_authenticated and (app is None or app.user_id is not None): + scan.user = request.user + if app is not None: + scan.app = app scan.status = 'In Progress' scan.progress = 1 scan.save() + if scan.user_id is None: + grant_guest_scan_access(request, scan) task_id = task_create_scan.delay(scan.id) scan.task = task_id.id scan.save() messages.success(request, 'Form submission successful') return redirect(reverse('scan', kwargs={"id": scan.id})) else: + form_class = ScanForm if request.user.is_authenticated else GuestScanForm if (app_id == ''): - form = ScanForm() + form = form_class() else: - app = Application.objects.get(pk=app_id) - form = ScanForm(initial={'app': app}) - if (settings.DEFECTDOJO_ENABLED == False): + form = form_class(initial={'app': app}) + if (settings.DEFECTDOJO_ENABLED == False) and 'defectdojo_id' in form.fields: form.fields.pop('defectdojo_id') return render(request, 'create_scan.html', { 'form': form, + 'guest_scan': not request.user.is_authenticated, }) + + +def scan_state(request, id): + scan = get_object_or_404(Scan, pk=id) + if not can_access_scan(request, scan): + raise Http404 + return get_scan_state(request, id) @login_required def delete_scan(request, scan_id=''): if request.method == 'POST': @@ -206,10 +240,14 @@ def delete_scan(request, scan_id=''): messages.warning(request, 'Removed successfully') return redirect('home') -@login_required def app(request, id): - app = Application.objects.get(pk=id) - scans = Scan.objects.filter(app=app.id).order_by('id') + app = get_object_or_404(Application, pk=id) + if not can_access_app(request, app): + raise Http404 + scans = Scan.objects.filter(app=app.id) + if app.user_id is None: + scans = scans.filter(user__isnull=True, pk__in=guest_scan_ids(request)) + scans = scans.order_by('id') scans_data = {} chart_labels = [] chart_data = [] @@ -233,19 +271,22 @@ def app(request, id): 'settings': settings, }) -@login_required def create_app(request): if request.method == 'POST': form = ApplicationForm(request.POST) if form.is_valid(): obj = form.save(commit=False) - obj.user = request.user - form_saved = obj.save() + if request.user.is_authenticated: + obj.user = request.user + obj.save() + if not request.user.is_authenticated: + grant_guest_app_access(request, obj) return redirect(reverse('create_scan', kwargs={"app_id": obj.id})) else: form = ApplicationForm() return render(request, 'create_app.html', { 'form': form, + 'guest_app': not request.user.is_authenticated, }) @login_required @@ -406,7 +447,9 @@ def malware(request): @login_required def update_virustotal(request, scan_id): - scan = Scan.objects.get(pk=scan_id) + scan = get_object_or_404(Scan, pk=scan_id) + if not can_access_scan(request, scan): + raise Http404 get_report_virus_total(scan, scan.sha256) return redirect(reverse('scan', kwargs={"id": scan_id})) @@ -415,7 +458,9 @@ def append_pdf(pdf, output): @login_required def export(request, id): - scan = Scan.objects.get(pk=id) + scan = get_object_or_404(Scan, pk=id) + if not can_access_scan(request, scan): + raise Http404 t = get_template('export.html') certificates = Certificate.objects.filter(scan=id) permissions = Permission.objects.filter(scan=id) @@ -464,4 +509,4 @@ def export(request, id): response = HttpResponse(pdf, content_type='application/pdf') response['Content-Disposition'] = "attachment; filename = scan.pdf" - return response \ No newline at end of file + return response