senaduka/old-riskletpy
1
0
Fork 0
Code Issues 5 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Added celerity background process

Browse Source
This commit is contained in:
Amir
2025-02-14 19:39:23 +01:00
parent 595b7a2a17
commit 65716ff842
5 changed files with 83 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/__init__.py
View File

@@ -0,0 +1,3 @@
from .celery import app as celery_app
__all__ = ('celery_app',)

52
backend/accounts/tasks.py Normal file
View File

@@ -0,0 +1,52 @@
from celery import shared_task
from backend.core.models import Organization, Document, Risk, Control, DocumentRiskControl
from backend.core.utils import get_top_risk, get_controls_for_risk
from django.shortcuts import get_object_or_404, render
@shared_task
def create_document_for_organization(confirmation_email):
organization = get_object_or_404(Organization, email=confirmation_email)
top_risk_ids = get_top_risk(organization)
top_risks = Risk.objects.filter(risk_id__in=top_risk_ids)
organization.risks.set(top_risks)
document = Document.objects.create(organization=organization)
document.add_segment('h1', "Top 10 Risks Identified")
risk_content = "\n\n".join([
f"Risk: {risk.risk_id} - {risk.risk_name} \n"
f"Category: {risk.category}\n"
f"Primary Impact: {risk.primary_impact} \n"
f"Secondary Impact: {risk.secondary_impact}\n"
f"Tertiary Impact: {risk.tretiary_impact} \n"
f"Detection Difficulty: {risk.detection_difficulty} \n"
f"Recovery Complexity: {risk.recovery_complexity} \n"
f"Business Impact Severity: {risk.businnes_impact_severity}\n"
for risk in top_risks
])
document.add_segment('body', f"Identified Risks: \n\n{risk_content}")
controls_content = "Mitigation Controls:\n\n"
for risk in top_risks:
controls_content += f"Risk: {risk.risk_id} - {risk.risk_name}\n"
selected_controls = get_controls_for_risk(risk)
for control_id, weight in selected_controls:
control = Control.objects.filter(id=control_id).first()
if control:
DocumentRiskControl.objects.create(
document=document,
risk=risk,
control=control,
weight=weight
)
controls_content += f" - Control: {control.name} (Impact Weight: {weight}/10)\n"
controls_content += "\n"
document.add_segment('body', controls_content)

62
backend/accounts/views.py
View File

@@ -5,23 +5,7 @@ from .models import EmailConfirmation
from django.shortcuts import get_object_or_404, render
from django.http import HttpResponse
from backend.accounts.utils import send_confirmation_email
from backend.core.utils import get_controls_for_risk, get_top_risk
from backend.core.models import Organization,Risk,Document,Control,DocumentRiskControl
class SignUpView(CreateView):
form_class = SignupForm
success_url = reverse_lazy('login')
template_name = 'accounts/signup.html'
from django.urls import reverse_lazy
from django.views.generic import CreateView
from django.shortcuts import get_object_or_404, render
from django.http import HttpResponse
from backend.accounts.forms import SignupForm
from backend.accounts.utils import send_confirmation_email
from backend.core.utils import get_controls_for_risk, get_top_risk
from backend.core.models import Organization, Risk, Document, Control, DocumentRiskControl
from .tasks import create_document_for_organization
class SignUpView(CreateView):
form_class = SignupForm
@@ -35,49 +19,9 @@ def confirm_email(request, uuid):
if confirmation.is_expired():
return render(request, 'confirmation_expired.html', {'email': confirmation.email})
organization = get_object_or_404(Organization, email=confirmation.email)
task = create_document_for_organization.delay(confirmation.email)
print(f"Task ID: {task.id}")
top_risk_ids = get_top_risk(organization)
top_risks = Risk.objects.filter(risk_id__in=top_risk_ids)
organization.risks.set(top_risks)
document = Document.objects.create(organization=organization)
document.add_segment('h1', "Top 10 Risks Identified")
risk_content = "\n\n".join([
f"Risk: {risk.risk_id} - {risk.risk_name} \n"
f"Category: {risk.category}\n"
f"Primary Impact: {risk.primary_impact} \n"
f"Secondary Impact: {risk.secondary_impact}\n"
f"Tertiary Impact: {risk.tretiary_impact} \n"
f"Detection Difficulty: {risk.detection_difficulty} \n"
f"Recovery Complexity: {risk.recovery_complexity} \n"
f"Business Impact Severity: {risk.businnes_impact_severity}\n"
for risk in top_risks
])
document.add_segment('body', f"Identified Risks: \n\n{risk_content}")
controls_content = "Mitigation Controls:\n\n"
for risk in top_risks:
controls_content += f"Risk: {risk.risk_id} - {risk.risk_name}\n"
selected_controls = get_controls_for_risk(risk)
for control_id, weight in selected_controls:
control = Control.objects.filter(id=control_id).first()
if control:
DocumentRiskControl.objects.create(
document=document,
risk=risk,
control=control,
weight=weight
)
controls_content += f" - Control: {control.name} (Impact Weight: {weight}/10)\n"
controls_content += "\n"
document.add_segment('body', controls_content)
return HttpResponse("Email is confirmed")

15
backend/celery.py Normal file
View File

@@ -0,0 +1,15 @@
# backend/celery.py
import os
from celery import Celery
# Set the default Django settings module
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'backend.settings')
# Create the Celery app
app = Celery('backend')
# Load configuration from Django settings
app.config_from_object('django.conf:settings', namespace='CELERY')
# Discover tasks in all installed apps
app.autodiscover_tasks()

10
backend/settings.py
View File

@@ -51,6 +51,8 @@ INSTALLED_APPS = [
'django_extensions',
'widget_tweaks',
'django_seed',
'django_celery_results',
# my apps
'backend.accounts.apps.AccountsConfig',
'backend.core.apps.CoreConfig',
@@ -156,3 +158,11 @@ DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'
LOGIN_URL = '/admin/login/'
LOGIN_REDIRECT_URL = 'core:index'
# LOGOUT_REDIRECT_URL = 'core:index'
# Celery Configuration
CELERY_BROKER_URL = 'redis://localhost:6380/0'
CELERY_RESULT_BACKEND = 'django-db' # Store task results in the Django database
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_TIMEZONE = 'UTC'