Swaped from CIS to NIST controls
This commit is contained in:
@@ -47,8 +47,8 @@ def create_document_for_organization(confirmation_email):
|
||||
weight=weight,
|
||||
likelihood=likelihood
|
||||
)
|
||||
controls_content += f" - Control: {control.name} (Impact Weight: {weight}/10) (Likelihood: {likelihood}/10)\n"
|
||||
|
||||
label = f"{control.subcategory} - {control.function or ''}".rstrip(" -")
|
||||
controls_content += f" - Control: {label} (Impact Weight: {weight}/10) (Likelihood: {likelihood}/10)\n"
|
||||
controls_content += "\n"
|
||||
|
||||
document.add_segment('body', controls_content)
|
||||
|
||||
@@ -32,7 +32,7 @@ class CeleryTaskTests(TestCase):
|
||||
integration_level="Highly Integrated"
|
||||
)
|
||||
self.risk = Risk.objects.create(risk_id="1", risk_name="Test Risk", category="Category1", primary_impact="High")
|
||||
self.control = Control.objects.create(name="Test Control")
|
||||
self.control = Control.objects.create(subcategory="PR.AA-01", function="Identity Management")
|
||||
|
||||
@patch("backend.accounts.tasks.get_top_risk")
|
||||
@patch("backend.accounts.tasks.get_controls_for_risk")
|
||||
|
||||
@@ -39,7 +39,8 @@ class RiskAdmin(admin.ModelAdmin):
|
||||
list_display = ['risk_id','risk_name','category']
|
||||
|
||||
class ControlAdmin(admin.ModelAdmin):
|
||||
list_display = ('id','safeguard_id','name', 'description')
|
||||
list_display = ('id', 'subcategory', 'function', 'category')
|
||||
search_fields = ('subcategory', 'function', 'category')
|
||||
|
||||
class DocumentRiskControlAdmin(admin.ModelAdmin):
|
||||
list_display = ('document', 'risk', 'control', 'weight','likelihood')
|
||||
|
||||
@@ -11,15 +11,19 @@ class Command(BaseCommand):
|
||||
def handle(self, *args, **options):
|
||||
csv_file_path = options["csv_file"]
|
||||
|
||||
with open(csv_file_path, mode="w", newline="", encoding="utf-8") as csv_file:
|
||||
fieldnames = ["CIS v8.1 Safeguards (Sub-Controls)"]
|
||||
with open(csv_file_path, mode="w", encoding="utf-8", newline="") as csv_file:
|
||||
fieldnames = ["Subcategory","Function","Category","Implementation_Examples","Effectiveness_Monitoring_Examples","Documentation_Score","Implementation_Score"]
|
||||
writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
|
||||
|
||||
writer.writeheader()
|
||||
|
||||
for control in Control.objects.all():
|
||||
for control in Control.objects.all().order_by("subcategory"):
|
||||
writer.writerow({
|
||||
"CIS v8.1 Safeguards (Sub-Controls)": control.name,
|
||||
"Subcategory": control.subcategory,
|
||||
"Function": control.function or "",
|
||||
"Category": control.category or "",
|
||||
"Implementation_Examples": control.implementation_examples or "",
|
||||
"Effectiveness_Monitoring_Examples": control.effectiveness_monitoring_examples or "",
|
||||
"Documentation_Score": control.documentation_score if control.documentation_score is not None else "",
|
||||
"Implementation_Score": control.implementation_score if control.implementation_score is not None else "",
|
||||
})
|
||||
|
||||
self.stdout.write(self.style.SUCCESS(f"Controls exported successfully to {csv_file_path}"))
|
||||
|
||||
@@ -13,16 +13,17 @@ class Command(BaseCommand):
|
||||
|
||||
with open(csv_file_path, mode="r", encoding="utf-8") as csv_file:
|
||||
reader = csv.DictReader(csv_file)
|
||||
|
||||
for row in reader:
|
||||
safeguard_id = row["Safeguard ID"].strip()
|
||||
safeguard = row["Name"].strip()
|
||||
description = row["Description"].strip()
|
||||
subcategory = (row.get("Subcategory") or "").strip()
|
||||
Control.objects.update_or_create(
|
||||
name=safeguard,
|
||||
safeguard_id = safeguard_id,
|
||||
description=description,
|
||||
defaults={"name": safeguard},
|
||||
subcategory=subcategory,
|
||||
defaults={
|
||||
"function": (row.get("Function") or "").strip() or None,
|
||||
"category": (row.get("Category") or "").strip() or None,
|
||||
"implementation_examples": (row.get("Implementation_Examples") or "").strip() or None,
|
||||
"effectiveness_monitoring_examples": (row.get("Effectiveness_Monitoring_Examples") or "").strip() or None,
|
||||
"documentation_score": int(row["Documentation_Score"]) if (row.get("Documentation_Score") or "").strip() else None,
|
||||
"implementation_score": int(row["Implementation_Score"]) if (row.get("Implementation_Score") or "").strip() else None,
|
||||
},
|
||||
)
|
||||
|
||||
self.stdout.write(self.style.SUCCESS("Safeguards imported successfully!"))
|
||||
self.stdout.write(self.style.SUCCESS("NIST controls imported successfully!"))
|
||||
|
||||
@@ -6,8 +6,8 @@ from backend.core.models import Control
|
||||
|
||||
class ExportControlsCommandTest(TestCase):
|
||||
def setUp(self):
|
||||
Control.objects.create(name="Test Safeguard 1")
|
||||
Control.objects.create(name="Test Safeguard 2")
|
||||
Control.objects.create(subcategory="PR.AA-01", function="Identity Management")
|
||||
Control.objects.create(subcategory="PR.DS-11", function="Backups")
|
||||
|
||||
self.csv_file_path = 'test_export_controls.csv'
|
||||
|
||||
@@ -26,5 +26,7 @@ class ExportControlsCommandTest(TestCase):
|
||||
|
||||
self.assertEqual(len(rows), 2)
|
||||
|
||||
self.assertEqual(rows[0]["CIS v8.1 Safeguards (Sub-Controls)"], "Test Safeguard 1")
|
||||
self.assertEqual(rows[1]["CIS v8.1 Safeguards (Sub-Controls)"], "Test Safeguard 2")
|
||||
self.assertIn("Subcategory", reader.fieldnames)
|
||||
self.assertIn("Function", reader.fieldnames)
|
||||
self.assertEqual(rows[0]["Subcategory"], "PR.AA-01")
|
||||
self.assertEqual(rows[1]["Subcategory"], "PR.DS-11")
|
||||
@@ -7,11 +7,21 @@ from backend.core.models import Control
|
||||
class ImportControlsCommandTest(TestCase):
|
||||
def setUp(self):
|
||||
self.csv_file_path = 'test_import_controls.csv'
|
||||
with open(self.csv_file_path, mode='w', encoding='utf-8') as csv_file:
|
||||
writer = csv.DictWriter(csv_file, fieldnames=["CIS v8.1 Safeguards (Sub-Controls)"])
|
||||
with open(self.csv_file_path, mode='w', encoding='utf-8', newline='') as csv_file:
|
||||
writer = csv.DictWriter(csv_file, fieldnames=[
|
||||
"Subcategory","Function","Category",
|
||||
"Implementation_Examples","Effectiveness_Monitoring_Examples",
|
||||
"Documentation_Score","Implementation_Score",
|
||||
])
|
||||
writer.writeheader()
|
||||
writer.writerow({"CIS v8.1 Safeguards (Sub-Controls)": "Test Safeguard 1"})
|
||||
writer.writerow({"CIS v8.1 Safeguards (Sub-Controls)": "Test Safeguard 2"})
|
||||
writer.writerow({
|
||||
"Subcategory":"GV.SC-06",
|
||||
"Function":"GOVERN (GV): ...",
|
||||
"Category":"Cybersecurity Supply Chain Risk Management (GV.SC)",
|
||||
"Implementation_Examples":"Ex1: ...",
|
||||
"Effectiveness_Monitoring_Examples":"",
|
||||
"Documentation_Score":"", "Implementation_Score":""
|
||||
})
|
||||
|
||||
def tearDown(self):
|
||||
if os.path.exists(self.csv_file_path):
|
||||
@@ -22,19 +32,17 @@ class ImportControlsCommandTest(TestCase):
|
||||
|
||||
call_command('import_controls', self.csv_file_path)
|
||||
|
||||
self.assertEqual(Control.objects.count(), 2)
|
||||
self.assertEqual(Control.objects.count(), 1)
|
||||
|
||||
safeguards = Control.objects.values_list('name', flat=True)
|
||||
self.assertIn("Test Safeguard 1", safeguards)
|
||||
self.assertIn("Test Safeguard 2", safeguards)
|
||||
controls = Control.objects.all()
|
||||
self.assertEqual(controls[0].subcategory, "GV.SC-06")
|
||||
|
||||
def test_import_controls_update(self):
|
||||
Control.objects.create(name="Test Safeguard 1")
|
||||
Control.objects.create(subcategory="GV.SC-06")
|
||||
|
||||
call_command('import_controls', self.csv_file_path)
|
||||
|
||||
self.assertEqual(Control.objects.count(), 2)
|
||||
self.assertEqual(Control.objects.count(), 1)
|
||||
|
||||
safeguards = Control.objects.values_list('name', flat=True)
|
||||
self.assertIn("Test Safeguard 1", safeguards)
|
||||
self.assertIn("Test Safeguard 2", safeguards)
|
||||
control = Control.objects.first()
|
||||
self.assertEqual(control.subcategory, "GV.SC-06")
|
||||
@@ -0,0 +1,67 @@
|
||||
# Generated by Django 5.1.3 on 2025-08-13 01:37
|
||||
|
||||
import django.utils.timezone
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('core', '0019_document_recomendations'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.RemoveField(
|
||||
model_name='control',
|
||||
name='description',
|
||||
),
|
||||
migrations.RemoveField(
|
||||
model_name='control',
|
||||
name='name',
|
||||
),
|
||||
migrations.RemoveField(
|
||||
model_name='control',
|
||||
name='safeguard_id',
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='control',
|
||||
name='category',
|
||||
field=models.TextField(blank=True, help_text='Category', null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='control',
|
||||
name='documentation_score',
|
||||
field=models.IntegerField(blank=True, help_text='Documentation Score', null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='control',
|
||||
name='effectiveness_monitoring_examples',
|
||||
field=models.TextField(blank=True, help_text='Effectiveness Monitoring Examples', null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='control',
|
||||
name='function',
|
||||
field=models.TextField(blank=True, help_text='Function', null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='control',
|
||||
name='implementation_examples',
|
||||
field=models.TextField(blank=True, help_text='Implementation Examples', null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='control',
|
||||
name='implementation_score',
|
||||
field=models.IntegerField(blank=True, help_text='Implementation Score', null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='control',
|
||||
name='subcategory',
|
||||
field=models.CharField(default=django.utils.timezone.now, help_text='NIST Subcategory', max_length=64, unique=True),
|
||||
preserve_default=False,
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='control',
|
||||
name='subcategory_description',
|
||||
field=models.TextField(blank=True, help_text='Subcategory Description', null=True),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,17 @@
|
||||
# Generated by Django 5.1.3 on 2025-08-14
|
||||
|
||||
from django.db import migrations
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('core', '0020_remove_control_description_remove_control_name_and_more'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.RemoveField(
|
||||
model_name='control',
|
||||
name='subcategory_description',
|
||||
),
|
||||
]
|
||||
@@ -159,11 +159,17 @@ class Risk(models.Model):
|
||||
|
||||
class Control(models.Model):
|
||||
id = models.AutoField(primary_key=True)
|
||||
safeguard_id = models.CharField(max_length=10, unique=True, null=True, help_text="Unique identifier for the safeguard (e.g. '1.1', '4.6')")
|
||||
name = models.CharField(max_length=255)
|
||||
description = models.TextField(default=" ", help_text="Description of the control")
|
||||
|
||||
subcategory = models.CharField(max_length=64, unique=True, null=False, blank=False, help_text='NIST Subcategory')
|
||||
function = models.TextField(null=True, blank=True, help_text="Function")
|
||||
category = models.TextField(null=True, blank=True,help_text="Category")
|
||||
implementation_examples = models.TextField(null=True, blank=True, help_text="Implementation Examples")
|
||||
effectiveness_monitoring_examples = models.TextField(null=True, blank=True, help_text="Effectiveness Monitoring Examples")
|
||||
documentation_score = models.IntegerField(null=True, blank=True, help_text="Documentation Score")
|
||||
implementation_score = models.IntegerField(null=True, blank=True, help_text="Implementation Score")
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.id} ({self.name})"
|
||||
return f"{self.id} ({self.subcategory} - {self.function or self.category or ''})".rstrip(" -() ")
|
||||
|
||||
class DocumentRiskControl(models.Model):
|
||||
document = models.ForeignKey(Document, on_delete=models.CASCADE)
|
||||
|
||||
@@ -52,7 +52,7 @@ def get_risk_table(document):
|
||||
controls = (
|
||||
DocumentRiskControl.objects
|
||||
.filter(document=document, risk_id=risk['id'])
|
||||
.values('control', 'control__name', 'weight', 'likelihood')
|
||||
.values('control', 'control__subcategory', 'control__function', 'weight', 'likelihood')
|
||||
.distinct()
|
||||
)
|
||||
max_weight = 10*5
|
||||
@@ -90,10 +90,12 @@ def get_safeguard_summary_table(risks_with_controls):
|
||||
for risk in risks_with_controls:
|
||||
for control in risk.get('controls', []):
|
||||
control_id = control.get('control')
|
||||
control_name = control.get('control__name')
|
||||
subc = control.get('control__subcategory') or ''
|
||||
func = control.get('control__function') or ''
|
||||
label = f"{subc} - {func}".rstrip(" -")
|
||||
if control_id:
|
||||
safeguard_counter[control_id] += 1
|
||||
safeguard_names[control_id] = control_name
|
||||
safeguard_names[control_id] = label
|
||||
|
||||
summary = []
|
||||
controls = Control.objects.filter(id__in=safeguard_counter.keys())
|
||||
@@ -103,9 +105,10 @@ def get_safeguard_summary_table(risks_with_controls):
|
||||
control = controls_map.get(control_id)
|
||||
summary.append({
|
||||
'id': control_id,
|
||||
'safeguard_id': control.safeguard_id if control else '',
|
||||
'subcategory': control.subcategory if control else '',
|
||||
'category': control.category if control else '',
|
||||
'function': control.function if control else '',
|
||||
'name': safeguard_names.get(control_id, ''),
|
||||
'description': control.description if control else '',
|
||||
'count': count,
|
||||
})
|
||||
summary.sort(key=lambda x: x['count'], reverse=True)
|
||||
|
||||
@@ -45,7 +45,7 @@ class UtilsTests(TestCase):
|
||||
primary_impact="Financial"
|
||||
)
|
||||
|
||||
self.controls = [Control.objects.create(id=i, name=f"Control {i}") for i in range(1, 11)]
|
||||
self.controls = [Control.objects.create(id=i, subcategory=f"C-{i}", function=f"Control {i}") for i in range(1, 11)]
|
||||
|
||||
def test_extract_organization_details(self):
|
||||
details = extract_organization_details(self.organization)
|
||||
@@ -148,25 +148,25 @@ class UtilsTests(TestCase):
|
||||
self.assertIsInstance(graph_data, str)
|
||||
self.assertTrue(len(graph_data) > 1000)
|
||||
|
||||
def test_get_safeguard_summary_table_basic(self):
|
||||
from backend.core.tables import get_safeguard_summary_table
|
||||
risks_with_controls = [
|
||||
{
|
||||
'risk': {'id': 1, 'name': 'Risk 1'},
|
||||
'controls': [
|
||||
{'control': 101, 'control__name': 'Control A'},
|
||||
{'control': 102, 'control__name': 'Control B'},
|
||||
]
|
||||
},
|
||||
{
|
||||
'risk': {'id': 2, 'name': 'Risk 2'},
|
||||
'controls': [
|
||||
{'control': 101, 'control__name': 'Control A'},
|
||||
]
|
||||
}
|
||||
]
|
||||
summary = get_safeguard_summary_table(risks_with_controls)
|
||||
self.assertEqual(summary, [
|
||||
{'id': 101, 'name': 'Control A', 'count': 2},
|
||||
{'id': 102, 'name': 'Control B', 'count': 1},
|
||||
])
|
||||
def test_get_safeguard_summary_table_basic(self):
|
||||
from backend.core.tables import get_safeguard_summary_table
|
||||
risks_with_controls = [
|
||||
{
|
||||
'risk': {'id': 1, 'name': 'Risk 1'},
|
||||
'controls': [
|
||||
{'control': 101, 'control__subcategory': 'PR.AA-01', 'control__function': 'Identity'},
|
||||
{'control': 102, 'control__subcategory': 'PR.DS-11', 'control__function': 'Backups'},
|
||||
]
|
||||
},
|
||||
{
|
||||
'risk': {'id': 2, 'name': 'Risk 2'},
|
||||
'controls': [
|
||||
{'control': 101, 'control__subcategory': 'PR.AA-01', 'control__function': 'Identity'},
|
||||
]
|
||||
}
|
||||
]
|
||||
summary = get_safeguard_summary_table(risks_with_controls)
|
||||
self.assertEqual(summary, [
|
||||
{'id': 101, 'subcategory': '', 'category': '', 'function': '', 'name': 'PR.AA-01 - Identity', 'count': 2},
|
||||
{'id': 102, 'subcategory': '', 'category': '', 'function': '', 'name': 'PR.DS-11 - Backups', 'count': 1},
|
||||
])
|
||||
@@ -44,8 +44,8 @@ class DocumentViewTest(TestCase):
|
||||
|
||||
self.risk1 = Risk.objects.create(risk_id=1, risk_name="Risk 1")
|
||||
self.risk2 = Risk.objects.create(risk_id=2, risk_name="Risk 2")
|
||||
self.control1 = Control.objects.create(id=1, name="Control A")
|
||||
self.control2 = Control.objects.create(id=2, name="Control B")
|
||||
self.control1 = Control.objects.create(id=1, subcategory="PR.AA-01", function="Identity Management")
|
||||
self.control2 = Control.objects.create(id=2, subcategory="PR.DS-11", function="Backups")
|
||||
|
||||
DocumentRiskControl.objects.create(id=1, document=self.document, risk=self.risk1, control=self.control1, weight=5, likelihood=3)
|
||||
DocumentRiskControl.objects.create(id=2, document=self.document, risk=self.risk1, control=self.control2, weight=7, likelihood=4)
|
||||
@@ -98,7 +98,7 @@ class DocumentViewTest(TestCase):
|
||||
<h4>Mitigating Controls:</h4>
|
||||
{% for control in item.controls %}
|
||||
<div class="control">
|
||||
<span class="name">{{ control.control__name }}</span> -
|
||||
<span class="name">{{ control.control__subcategory }} - {{ control.control__function }}</span> -
|
||||
<span class="weight">Weight: {{ control.weight }}</span>
|
||||
</div>
|
||||
{% endfor %}
|
||||
|
||||
@@ -76,10 +76,10 @@ def get_controls_for_risk(risk, organization):
|
||||
client = OpenAI(api_key=settings.OPENAI_API_KEY)
|
||||
all_controls = Control.objects.all()
|
||||
organization_details = extract_organization_details(organization)
|
||||
control_list = [f"Control ID: {control.id}, Control Name: {control.name}" for control in all_controls]
|
||||
control_list = [f"Control ID: {control.id}, Control Name: {control.subcategory} - {control.function or ''}".rstrip(" -") for control in all_controls]
|
||||
valid_control_ids = {control.id for control in all_controls}
|
||||
control_map = {control.id: control.name for control in all_controls}
|
||||
|
||||
control_map = {control.id: (f"{control.subcategory} - {control.function or ''}").rstrip(" -") for control in all_controls}
|
||||
|
||||
def fetch_controls(prompt):
|
||||
response = client.chat.completions.create(
|
||||
model="gpt-4o-mini",
|
||||
@@ -229,13 +229,13 @@ def generate_recommendations(risks_with_controls, organization):
|
||||
risk: id, name, category, risk_description (or similar)
|
||||
r_impact (inherent impact 1–5), r_likelihood (inherent likelihood 1–5), risk_score
|
||||
residual_impact, residual_likelihood, residual_risk_score (may be present)
|
||||
controls: list of controls, each with control__name, weight (1–5 effectiveness), likelihood (1–5 occurrence modifier)
|
||||
controls: list of controls, each with control__subcategory and control__function, weight (1–5 effectiveness), likelihood (1–5 occurrence modifier)
|
||||
|
||||
Task:
|
||||
1) Compute a priority score per control = weight × likelihood. Aggregate scores across all risks and cluster into 3–5 thematic areas that best match the actual controls and risk names (e.g., Access Control & MFA, Patch & Vulnerability Management, Vendor/Third-Party Risk Management, Network Security & Segmentation, Logging/Monitoring/Detection, Incident Response & BCDR, Ransomware Prevention & Recovery, Cryptography & Key Management). Do not invent themes without support in the inputs.
|
||||
2) For each chosen theme, produce 3–5 concrete actions derived from the highest-priority controls. Tailor to the organization_details where appropriate. Prefer steps that reduce both likelihood and impact.
|
||||
3) Each bullet should be 1–2 sentences: start with a clear, imperative recommendation, and (optionally) add a brief explanation or context. Still keep it concise and actionable.
|
||||
4) Use only the control__name for reference—do NOT include or reference control IDs, years (e.g., 2024), or quarter references (Q1, Q2, Q3, Q4) anywhere in the output.
|
||||
4) Use only the control label (i.e., "subcategory - function") for reference—do NOT include or reference control IDs, years, or quarter references (Q1, Q2, Q3, Q4).
|
||||
5) Do not introduce controls that are not represented in the provided controls list.
|
||||
|
||||
Output format (STRICT):
|
||||
|
||||
Reference in New Issue
Block a user