from datetime import datetime import markdown from django.contrib import messages from django.db.models import Count from django.http import HttpResponse, HttpResponseRedirect, JsonResponse from django.shortcuts import get_object_or_404, render from django.template.loader import get_template from django.urls import reverse from django.utils import timezone from django_celery_beat.models import (ClockedSchedule, IntervalSchedule, PeriodicTask) from reNgine.celery import app from reNgine.common_func import * from reNgine.definitions import ABORTED_TASK, SUCCESS_TASK from reNgine.tasks import create_scan_activity, initiate_scan, run_command from scanEngine.models import EngineType from startScan.models import * from targetApp.models import * from weasyprint import HTML def scan_history(request): host = ScanHistory.objects.all().order_by('-start_scan_date') context = {'scan_history_active': 'active', "scan_history": host} return render(request, 'startScan/history.html', context) def subscan_history(request): subscans = SubScan.objects.all().order_by('-start_scan_date') context = {'scan_history_active': 'active', "subscans": subscans} return render(request, 'startScan/subscan_history.html', context) def detail_scan(request, id=None): ctx = {} if not id: return render(request, 'startScan/detail_scan.html', ctx) # Get scan objects scan = get_object_or_404(ScanHistory, id=id) domain_id = scan.domain.id scan_engines = EngineType.objects.order_by('engine_name').all() recent_scans = ScanHistory.objects.filter(domain__id=domain_id) last_scans = ( ScanHistory.objects .filter(domain__id=domain_id) .filter(tasks__overlap=['subdomain_discovery']) .filter(id__lte=id) .filter(scan_status=2) ) # Get all kind of objects associated with our ScanHistory object emails = Email.objects.filter(emails__in=[scan]) employees = Employee.objects.filter(employees__in=[scan]) subdomains = Subdomain.objects.filter(scan_history=scan) endpoints = EndPoint.objects.filter(scan_history=scan) vulns = Vulnerability.objects.filter(scan_history=scan) vulns_tags = VulnerabilityTags.objects.filter(vuln_tags__in=vulns) ip_addresses = IpAddress.objects.filter(ip_addresses__in=subdomains) geo_isos = CountryISO.objects.filter(ipaddress__in=ip_addresses) scan_activity = ScanActivity.objects.filter(scan_of__id=id).order_by('time') cves = CveId.objects.filter(cve_ids__in=vulns) cwes = CweId.objects.filter(cwe_ids__in=vulns) # HTTP statuses http_statuses = ( subdomains .exclude(http_status=0) .values('http_status') .annotate(Count('http_status')) ) # CVEs / CWes common_cves = ( cves .annotate(nused=Count('cve_ids')) .order_by('-nused') .values('name', 'nused') [:10] ) common_cwes = ( cwes .annotate(nused=Count('cwe_ids')) .order_by('-nused') .values('name', 'nused') [:10] ) # Tags common_tags = ( vulns_tags .annotate(nused=Count('vuln_tags')) .order_by('-nused') .values('name', 'nused') [:7] ) # Countries asset_countries = ( geo_isos .annotate(count=Count('iso')) .order_by('-count') ) # Subdomains subdomain_count = ( subdomains .values('name') .distinct() .count() ) alive_count = ( subdomains .values('name') .distinct() .filter(http_status__exact=200) .count() ) important_count = ( subdomains .values('name') .distinct() .filter(is_important=True) .count() ) # Endpoints endpoint_count = ( endpoints .values('http_url') .distinct() .count() ) endpoint_alive_count = ( endpoints .filter(http_status__exact=200) # TODO: use is_alive() func as it's more precise .values('http_url') .distinct() .count() ) # Vulnerabilities common_vulns = ( vulns .exclude(severity=0) .values('name', 'severity') .annotate(count=Count('name')) .order_by('-count') [:10] ) info_count = vulns.filter(severity=0).count() low_count = vulns.filter(severity=1).count() medium_count = vulns.filter(severity=2).count() high_count = vulns.filter(severity=3).count() critical_count = vulns.filter(severity=4).count() unknown_count = vulns.filter(severity=-1).count() total_count = vulns.count() total_count_ignore_info = vulns.exclude(severity=0).count() # Emails exposed_count = emails.exclude(password__isnull=True).count() # Build render context ctx = { 'scan_history_id': id, 'history': scan, 'scan_activity': scan_activity, 'subdomain_count': subdomain_count, 'alive_count': alive_count, 'important_count': important_count, 'endpoint_count': endpoint_count, 'endpoint_alive_count': endpoint_alive_count, 'info_count': info_count, 'low_count': low_count, 'medium_count': medium_count, 'high_count': high_count, 'critical_count': critical_count, 'unknown_count': unknown_count, 'total_vulnerability_count': total_count, 'total_vul_ignore_info_count': total_count_ignore_info, 'vulnerability_list': vulns.order_by('-severity').all(), 'scan_history_active': 'active', 'scan_engines': scan_engines, 'exposed_count': exposed_count, 'email_count': emails.count(), 'employees_count': employees.count(), 'most_recent_scans': recent_scans.order_by('-start_scan_date')[:1], 'http_status_breakdown': http_statuses, 'most_common_cve': common_cves, 'most_common_cwe': common_cwes, 'most_common_tags': common_tags, 'most_common_vulnerability': common_vulns, 'asset_countries': asset_countries } # Find number of matched GF patterns if scan.used_gf_patterns: count_gf = {} for gf in scan.used_gf_patterns.split(','): count_gf[gf] = ( endpoints .filter(matched_gf_patterns__icontains=gf) .count() ) ctx['matched_gf_count'] = count_gf # Find last scan for this domain if last_scans.count() > 1: last_scan = last_scans.order_by('-start_scan_date')[1] ctx['last_scan'] = last_scan return render(request, 'startScan/detail_scan.html', ctx) def all_subdomains(request): subdomains = Subdomain.objects scan_engines = EngineType.objects.order_by('engine_name').all() alive_subdomains = subdomains.filter(http_status__exact=200) # TODO: replace this with is_alive() function important_subdomains = ( subdomains .filter(is_important=True) .values('name') .distinct() .count() ) context = { 'scan_history_id': id, 'scan_history_active': 'active', 'scan_engines': scan_engines, 'subdomain_count': subdomains.values('name').distinct().count(), 'alive_count': alive_subdomains.values('name').distinct().count(), 'important_count': important_subdomains } return render(request, 'startScan/subdomains.html', context) def detail_vuln_scan(request, id=None): if id: history = get_object_or_404(ScanHistory, id=id) context = {'scan_history_id': id, 'history': history} else: context = {'vuln_scan_active': 'true'} return render(request, 'startScan/vulnerabilities.html', context) def all_endpoints(request): context = { 'scan_history_active': 'active' } return render(request, 'startScan/endpoints.html', context) def start_scan_ui(request, domain_id): domain = get_object_or_404(Domain, id=domain_id) if request.method == "POST": # Get imported and out-of-scope subdomains subdomains_in = request.POST['importSubdomainTextArea'].split() subdomains_in = [s.rstrip() for s in subdomains_in if s] subdomains_out = request.POST['outOfScopeSubdomainTextarea'].split() subdomains_out = [s.rstrip() for s in subdomains_out if s] paths = request.POST['filterPath'].split() filterPath = [s.rstrip() for s in paths if s] if len(filterPath) > 0: filterPath = filterPath[0] else: filterPath = '' # Get engine type engine_id = request.POST['scan_mode'] # Create ScanHistory object scan_history_id = create_scan_object(domain_id, engine_id) scan = ScanHistory.objects.get(pk=scan_history_id) # Start the celery task kwargs = { 'scan_history_id': scan.id, 'domain_id': domain.id, 'engine_id': engine_id, 'scan_type': LIVE_SCAN, 'results_dir': '/usr/src/scan_results', 'imported_subdomains': subdomains_in, 'out_of_scope_subdomains': subdomains_out, 'url_filter': filterPath } initiate_scan.apply_async(kwargs=kwargs) scan.save() # Send start notif messages.add_message( request, messages.INFO, f'Scan Started for {domain.name}') return HttpResponseRedirect(reverse('scan_history')) # GET request engine = EngineType.objects.order_by('engine_name') custom_engine_count = ( EngineType.objects .filter(default_engine=False) .count() ) context = { 'scan_history_active': 'active', 'domain': domain, 'engines': engine, 'custom_engine_count': custom_engine_count} return render(request, 'startScan/start_scan_ui.html', context) def start_multiple_scan(request): # domain = get_object_or_404(Domain, id=host_id) if request.method == "POST": if request.POST.get('scan_mode', 0): # if scan mode is available, then start the scan # get engine type engine_id = request.POST['scan_mode'] list_of_domains = request.POST['list_of_domain_id'] for domain_id in list_of_domains.split(","): # Start the celery task scan_history_id = create_scan_object(domain_id, engine_id) kwargs = { 'scan_history_id': scan_history_id, 'domain_id': domain.id, 'engine_id': engine_id, 'scan_type': LIVE_SCAN, 'results_dir': '/usr/src/scan_results', # TODO: Add this to multiple scan view # 'imported_subdomains': subdomains_in, # 'out_of_scope_subdomains': subdomains_out } initiate_scan.apply_async(kwargs=kwargs) # Send start notif messages.add_message( request, messages.INFO, 'Scan Started for multiple targets') return HttpResponseRedirect(reverse('scan_history')) else: # this else condition will have post request from the scan page # containing all the targets id list_of_domain_name = [] list_of_domain_id = [] for key, value in request.POST.items(): if key != "list_target_table_length" and key != "csrfmiddlewaretoken": domain = get_object_or_404(Domain, id=value) list_of_domain_name.append(domain.name) list_of_domain_id.append(value) domain_ids = ",".join(list_of_domain_id) # GET request engines = EngineType.objects custom_engine_count = ( engines .filter(default_engine=False) .count() ) context = { 'scan_history_active': 'active', 'engines': engines, 'domain_list': list_of_domain_name, 'domain_ids': domain_ids, 'custom_engine_count': custom_engine_count } return render(request, 'startScan/start_multiple_scan_ui.html', context) def export_subdomains(request, scan_id): subdomain_list = Subdomain.objects.filter(scan_history__id=scan_id) scan = ScanHistory.objects.get(id=scan_id) response_body = "" for domain in subdomain_list: response_body += response_body + domain.name + "\n" scan_start_date_str = str(scan.start_scan_date.date()) domain_name = scan.domain.name response = HttpResponse(response_body, content_type='text/plain') response['Content-Disposition'] = ( f'attachment; filename="subdomains_{domain_name}_{scan_start_date_str}.txt"' ) return response def export_endpoints(request, scan_id): endpoint_list = EndPoint.objects.filter(scan_history__id=scan_id) scan = ScanHistory.objects.get(id=scan_id) response_body = "" for endpoint in endpoint_list: response_body += endpoint.http_url + "\n" scan_start_date_str = str(scan.start_scan_date.date()) domain_name = scan.domain.name response = HttpResponse(response_body, content_type='text/plain') response['Content-Disposition'] = ( f'attachment; filename="endpoints_{domain_name}_{scan_start_date_str}.txt"' ) return response def export_urls(request, scan_id): urls_list = Subdomain.objects.filter(scan_history__id=scan_id) scan = ScanHistory.objects.get(id=scan_id) response_body = "" for url in urls_list: if url.http_url: response_body += response_body + url.http_url + "\n" scan_start_date_str = str(scan.start_scan_date.date()) domain_name = scan.domain.name response = HttpResponse(response_body, content_type='text/plain') response['Content-Disposition'] = ( f'attachment; filename="urls_{domain_name}_{scan_start_date_str}.txt"' ) return response def delete_scan(request, id): obj = get_object_or_404(ScanHistory, id=id) if request.method == "POST": delete_dir = obj.results_dir run_command('rm -rf /usr/src/scan_results/' + delete_dir) obj.delete() messageData = {'status': 'true'} messages.add_message( request, messages.INFO, 'Scan history successfully deleted!' ) else: messageData = {'status': 'false'} messages.add_message( request, messages.INFO, 'Oops! something went wrong!' ) return JsonResponse(messageData) def stop_scan(request, id): if request.method == "POST": scan = get_object_or_404(ScanHistory, id=id) scan.scan_status = ABORTED_TASK scan.save() try: for task_id in scan.celery_ids: app.control.revoke(task_id, terminate=True, signal='SIGKILL') tasks = ( ScanActivity.objects .filter(scan_of=scan) .filter(status=RUNNING_TASK) .order_by('-pk') ) for task in tasks: task.status = ABORTED_TASK task.time = timezone.now() task.save() create_scan_activity(scan.id, "Scan aborted", SUCCESS_TASK) response = {'status': True} messages.add_message( request, messages.INFO, 'Scan successfully stopped!' ) except Exception as e: logger.error(e) response = {'status': False} messages.add_message( request, messages.ERROR, f'Scan failed to stop ! Error: {str(e)}' ) return JsonResponse(response) return scan_history(request) def schedule_scan(request, host_id): domain = Domain.objects.get(id=host_id) if request.method == "POST": scheduled_mode = request.POST['scheduled_mode'] engine_type = int(request.POST['scan_mode']) # Get imported and out-of-scope subdomains subdomains_in = request.POST['importSubdomainTextArea'].split() subdomains_in = [s.rstrip() for s in subdomains_in if s] subdomains_out = request.POST['outOfScopeSubdomainTextarea'].split() subdomains_out = [s.rstrip() for s in subdomains_out if s] # Get engine type engine = get_object_or_404(EngineType, id=engine_type) timestr = str(datetime.strftime(timezone.now(), '%Y_%m_%d_%H_%M_%S')) task_name = f'{engine.engine_name} for {domain.name}: {timestr}' if scheduled_mode == 'periodic': frequency_value = int(request.POST['frequency']) frequency_type = request.POST['frequency_type'] if frequency_type == 'minutes': period = IntervalSchedule.MINUTES elif frequency_type == 'hours': period = IntervalSchedule.HOURS elif frequency_type == 'days': period = IntervalSchedule.DAYS elif frequency_type == 'weeks': period = IntervalSchedule.DAYS frequency_value *= 7 elif frequency_type == 'months': period = IntervalSchedule.DAYS frequency_value *= 30 schedule, _ = IntervalSchedule.objects.get_or_create( every=frequency_value, period=period) kwargs = { 'domain_id': host_id, 'engine_id': engine.id, 'scan_history_id': 1, 'scan_type': SCHEDULED_SCAN, 'imported_subdomains': subdomains_in, 'out_of_scope_subdomains': subdomains_out } PeriodicTask.objects.create(interval=schedule, name=task_name, task='reNgine.tasks.initiate_scan', kwargs=json.dumps(kwargs)) elif scheduled_mode == 'clocked': schedule_time = request.POST['scheduled_time'] clock, _ = ClockedSchedule.objects.get_or_create( clocked_time=schedule_time) kwargs = { 'scan_history_id': 0, 'domain_id': host_id, 'engine_id': engine.id, 'scan_type': SCHEDULED_SCAN, 'imported_subdomains': subdomains_in, 'out_of_scope_subdomains': subdomains_out } PeriodicTask.objects.create(clocked=clock, one_off=True, name=task_name, task='reNgine.tasks.initiate_scan', kwargs=json.dumps(kwargs)) messages.add_message( request, messages.INFO, f'Scan Scheduled for {domain.name}' ) return HttpResponseRedirect(reverse('scheduled_scan_view')) # GET request engines = EngineType.objects custom_engine_count = ( engines .filter(default_engine=False) .count() ) context = { 'scan_history_active': 'active', 'domain': domain, 'engines': engines, 'custom_engine_count': custom_engine_count} return render(request, 'startScan/schedule_scan_ui.html', context) def scheduled_scan_view(request): scheduled_tasks = ( PeriodicTask.objects .all() .exclude(name='celery.backend_cleanup') ) context = { 'scheduled_scan_active': 'active', 'scheduled_tasks': scheduled_tasks, } return render(request, 'startScan/schedule_scan_list.html', context) def delete_scheduled_task(request, id): task_object = get_object_or_404(PeriodicTask, id=id) if request.method == "POST": task_object.delete() messageData = {'status': 'true'} messages.add_message( request, messages.INFO, 'Scheduled Scan successfully deleted!') else: messageData = {'status': 'false'} messages.add_message( request, messages.INFO, 'Oops! something went wrong!') return JsonResponse(messageData) def change_scheduled_task_status(request, id): if request.method == 'POST': task = PeriodicTask.objects.get(id=id) task.enabled = not task.enabled task.save() return HttpResponse('') def change_vuln_status(request, id): if request.method == 'POST': vuln = Vulnerability.objects.get(id=id) vuln.open_status = not vuln.open_status vuln.save() return HttpResponse('') def create_scan_object(host_id, engine_id): ''' create task with pending status so that celery task will execute when threads are free ''' # get current time current_scan_time = timezone.now() # fetch engine and domain object engine = EngineType.objects.get(pk=engine_id) domain = Domain.objects.get(pk=host_id) scan = ScanHistory() scan.scan_status = INITIATED_TASK scan.domain = domain scan.scan_type = engine scan.start_scan_date = current_scan_time scan.save() # save last scan date for domain model domain.start_scan_date = current_scan_time domain.save() return scan.id def delete_all_scan_results(request): if request.method == 'POST': ScanHistory.objects.all().delete() messageData = {'status': 'true'} messages.add_message( request, messages.INFO, 'All Scan History successfully deleted!') return JsonResponse(messageData) def delete_all_screenshots(request): if request.method == 'POST': run_command('rm -rf /usr/src/scan_results/*') messageData = {'status': 'true'} messages.add_message( request, messages.INFO, 'Screenshots successfully deleted!') return JsonResponse(messageData) def visualise(request, id): scan = ScanHistory.objects.get(id=id) context = { 'scan_id': id, 'scan_history': scan, } return render(request, 'startScan/visualise.html', context) def start_organization_scan(request, id): organization = get_object_or_404(Organization, id=id) if request.method == "POST": engine_id = request.POST['scan_mode'] # Start Celery task for each organization's domains for domain in organization.get_domains(): scan_history_id = create_scan_object(domain.id, engine_id) kwargs = { 'scan_history_id': scan_history_id, 'domain_id': domain.id, 'engine_id': engine_id, 'scan_type': LIVE_SCAN, 'results_dir': '/usr/src/scan_results', # TODO: Add this to multiple scan view # 'imported_subdomains': subdomains_in, # 'out_of_scope_subdomains': subdomains_out } initiate_scan.apply_async(kwargs=kwargs) # Send start notif ndomains = len(organization.get_domains()) messages.add_message( request, messages.INFO, f'Scan Started for {ndomains} domains in organization {organization.name}') return HttpResponseRedirect(reverse('scan_history')) # GET request engine = EngineType.objects.order_by('engine_name') custom_engine_count = EngineType.objects.filter(default_engine=False).count() domain_list = organization.get_domains() context = { 'organization_data_active': 'true', 'list_organization_li': 'active', 'organization': organization, 'engines': engine, 'domain_list': domain_list, 'custom_engine_count': custom_engine_count} return render(request, 'organization/start_scan.html', context) def schedule_organization_scan(request, id): organization =Organization.objects.get(id=id) if request.method == "POST": engine_type = int(request.POST['scan_mode']) engine = get_object_or_404(EngineType, id=engine_type) scheduled_mode = request.POST['scheduled_mode'] for domain in organization.get_domains(): timestr = str(datetime.strftime(timezone.now(), '%Y_%m_%d_%H_%M_%S')) task_name = f'{engine.engine_name} for {domain.name}: {timestr}' # Period task if scheduled_mode == 'periodic': frequency_value = int(request.POST['frequency']) frequency_type = request.POST['frequency_type'] if frequency_type == 'minutes': period = IntervalSchedule.MINUTES elif frequency_type == 'hours': period = IntervalSchedule.HOURS elif frequency_type == 'days': period = IntervalSchedule.DAYS elif frequency_type == 'weeks': period = IntervalSchedule.DAYS frequency_value *= 7 elif frequency_type == 'months': period = IntervalSchedule.DAYS frequency_value *= 30 schedule, _ = IntervalSchedule.objects.get_or_create( every=frequency_value, period=period ) _kwargs = json.dumps({ 'domain_id': domain.id, 'engine_id': engine.id, 'scan_history_id': 0, 'scan_type': SCHEDULED_SCAN, 'imported_subdomains': None }) PeriodicTask.objects.create( interval=schedule, name=task_name, task='reNgine.tasks.initiate_scan', kwargs=_kwargs ) # Clocked task elif scheduled_mode == 'clocked': schedule_time = request.POST['scheduled_time'] clock, _ = ClockedSchedule.objects.get_or_create( clocked_time=schedule_time ) _kwargs = json.dumps({ 'domain_id': domain.id, 'engine_id': engine.id, 'scan_history_id': 0, 'scan_type': LIVE_SCAN, 'imported_subdomains': None }) PeriodicTask.objects.create(clocked=clock, one_off=True, name=task_name, task='reNgine.tasks.initiate_scan', kwargs=_kwargs ) # Send start notif ndomains = len(organization.get_domains()) messages.add_message( request, messages.INFO, f'Scan started for {ndomains} domains in organization {organization.name}' ) return HttpResponseRedirect(reverse('scheduled_scan_view')) # GET request engine = EngineType.objects custom_engine_count = EngineType.objects.filter(default_engine=False).count() context = { 'scan_history_active': 'active', 'organization': organization, 'domain_list': organization.get_domains(), 'engines': engine, 'custom_engine_count': custom_engine_count } return render(request, 'organization/schedule_scan_ui.html', context) def delete_scans(request): if request.method == "POST": for key, value in request.POST.items(): if key == 'scan_history_table_length' or key == 'csrfmiddlewaretoken': continue scan = get_object_or_404(ScanHistory, id=value) delete_dir = scan.results_dir run_command('rm -rf /usr/src/scan_results/' + delete_dir) scan.delete() messages.add_message( request, messages.INFO, 'All Scans deleted!') return HttpResponseRedirect(reverse('scan_history')) def customize_report(request, id): scan = ScanHistory.objects.get(id=id) context = { 'scan_id': id, 'scan_history': scan, } return render(request, 'startScan/customize_report.html', context) def create_report(request, id): # primary_color = '#FFB74D' # secondary_color = '#212121' # get report type report_type = request.GET['report_type'] if 'report_type' in request.GET else 'full' if report_type == 'recon': show_recon = True show_vuln = False report_name = 'Reconnaissance Report' elif report_type == 'vulnerability': show_recon = False show_vuln = True report_name = 'Vulnerability Report' else: # default show_recon = True show_vuln = True report_name = 'Full Scan Report' scan = ScanHistory.objects.get(id=id) vulns = ( Vulnerability.objects .filter(scan_history=scan) .order_by('-severity') ) unique_vulns = ( Vulnerability.objects .filter(scan_history=scan) .values("name", "severity") .annotate(count=Count('name')) .order_by('-severity', '-count') ) subdomains = ( Subdomain.objects .filter(scan_history=scan) .order_by('-content_length') ) subdomain_alive_count = ( Subdomain.objects .filter(scan_history__id=id) .values('name') .distinct() .filter(http_status__exact=200) .count() ) interesting_subdomains = get_interesting_subdomains(scan_history=id) ip_addresses = ( IpAddress.objects .filter(ip_addresses__in=subdomains) .distinct() ) data = { 'scan_object': scan, 'unique_vulnerabilities': unique_vulns, 'vulns': vulns, 'subdomain_alive_count': subdomain_alive_count, 'interesting_subdomains': interesting_subdomains, 'subdomains': subdomains, 'ip_addresses': ip_addresses, 'show_recon': show_recon, 'show_vuln': show_vuln, 'report_name': report_name, } # Get report related config vuln_report_query = VulnerabilityReportSetting.objects.all() if vuln_report_query.exists(): report = vuln_report_query[0] data['company_name'] = report.company_name data['company_address'] = report.company_address data['company_email'] = report.company_email data['company_website'] = report.company_website data['show_rengine_banner'] = report.show_rengine_banner data['show_footer'] = report.show_footer data['footer_text'] = report.footer_text data['show_executive_summary'] = report.show_executive_summary # Replace executive_summary_description with template syntax description = report.executive_summary_description description = description.replace('{scan_date}', scan.start_scan_date.strftime('%d %B, %Y')) description = description.replace('{company_name}', report.company_name) description = description.replace('{target_name}', scan.domain.name) description = description.replace('{subdomain_count}', str(subdomains.count())) description = description.replace('{vulnerability_count}', str(vulns.count())) description = description.replace('{critical_count}', str(vulns.filter(severity=4).count())) description = description.replace('{high_count}', str(vulns.filter(severity=3).count())) description = description.replace('{medium_count}', str(vulns.filter(severity=2).count())) description = description.replace('{low_count}', str(vulns.filter(severity=1).count())) description = description.replace('{info_count}', str(vulns.filter(severity=0).count())) description = description.replace('{unknown_count}', str(vulns.filter(severity=-1).count())) if scan.domain.description: description = description.replace('{target_description}', scan.domain.description) # Convert to Markdown data['executive_summary_description'] = markdown.markdown(description) data['primary_color'] = report.primary_color data['secondary_color'] = report.secondary_color template = get_template('report/template.html') html = template.render(data) pdf = HTML(string=html).write_pdf() if 'download' in request.GET: response = HttpResponse(pdf, content_type='application/octet-stream') else: response = HttpResponse(pdf, content_type='application/pdf') return response