Develop aspen dynamic control info

CHANGELOG.md

@@ -27,6 +27,7 @@ Release adds support for private components and integrations with third party se
 * Display the control framework along side of controls in component control listing page.
 * Remove icons from project listing.
 * Add Component search filter to filter results to components owned by user.
+* Display project control text partially on page with modal for full text.
 
 **Developer changes**
 
@@ -49,6 +50,7 @@ Release adds support for private components and integrations with third party se
 * Assign owners to components imported via OSCAL. If no user is identified during component (element creation) assign first Superuser (administrator) as component owner.
 * Support navigating to specific tab on component library component page using URL hash (#) reference.
 * Protype integrations System Summary page.
+* Add Django admin command to create shell integration directory.
 * Refactor and OIDC authentication for proper testing of admin and not admin roles.
 * Create a new system via name given by a string in URL.
 * Add a large set of sample components (150+) generated from STIGs.

controls/management/commands/importcomponents.py

@@ -1,24 +1,15 @@
-import sys
 import os.path
-
 from django.core.management import call_command
-from django.core.management.base import BaseCommand, CommandError
-from django.db import transaction, models
-from django.db.utils import OperationalError
-from django.conf import settings
+from django.core.management.base import BaseCommand
 from pathlib import Path
-from pathlib import PurePath
-from django.utils.text import slugify
 
 # from siteapp.models import User, Organization, Portfolio
-from controls.models import Element, Statement
 # from controls.views import system_element_download_oscal_json
-from controls.views import OSCALComponentSerializer, ComponentImporter
-
-import fs, fs.errors
+from controls.views import ComponentImporter
 
 
 class Command(BaseCommand):
+    """Import directory of component files"""
     help = 'Import directory of component files.'
 
     def add_arguments(self, parser):
@@ -28,7 +19,6 @@ def add_arguments(self, parser):
         parser.add_argument('--stopinvalid', default=True, action='store_true')
         parser.add_argument('--no-stopinvalid', dest='stopinvalid', action='store_false')
 
-
     def handle(self, *args, **options):
 
         # Configure

controls/views.py

@@ -1,6 +1,5 @@
 import functools
 import logging
-import operator
 import pathlib
 import random
 import shutil
@@ -54,6 +53,7 @@
 from .utilities import *
 from siteapp.utils.views_helper import project_context
 from integrations.models import Integration
+from integrations.utils.integration import get_control_data_enhancements
 
 logging.basicConfig()
 import structlog
@@ -2298,7 +2298,8 @@ def editor(request, system_id, catalog_key, cl_id):
         # Define status options
         impl_statuses = ["Not implemented", "Planned", "Partially implemented", "Implemented", "Unknown"]
 
-      # Only elements for the given control id, sid, and statement type
+        # Only elements for the given control id, sid, and statement type
+        control_matrix = get_control_data_enhancements(request, catalog_key, cl_id)
 
         elements =  Element.objects.all().exclude(element_type='system')
 
@@ -2307,6 +2308,7 @@ def editor(request, system_id, catalog_key, cl_id):
             "project": project,
             "catalog": catalog,
             "control": cg_flat[cl_id.lower()],
+            "control_matrix": control_matrix,
             "impl_smts": impl_smts,
             "impl_statuses": impl_statuses,
             "impl_smts_legacy": impl_smts_legacy,
@@ -4072,7 +4074,7 @@ def system_summary_1_aspen(request, system_id):
     context = {
         "system": system_summary,
         #"project": project,
-        "project": projects,
+        "projects": projects,
         "system_events": system_events,
         # "deployments": deployments,
         "display_urls": project_context(project)

integrations/controlmatrix/README.md

@@ -0,0 +1,52 @@
+# ABOUT Integration Example
+
+## Configure
+
+### Step 1: Create an Integration record in Django admin:
+
+- Name: controlmatrix
+- Description: Integration to support Example service
+- Config:
+```json
+{
+    "base_url": "http://controlmatrix-test.agency.gov/controlmatrix/api",
+    "personal_access_token": "<personal_access_token>"
+}
+```
+- Config schema:
+```json
+{}
+```
+
+For local dev and testing, create an Integration record in Django admin for CSAM mock service:
+
+- Name: controlmatrix
+- Description: Integration to support Example service
+- Config:
+```json
+{
+    "base_url": "http://localhost:9009",
+    "personal_access_token": "FAD619"
+}
+```
+- Config schema:
+```json
+{}
+```
+
+### Step 2: Add route to `integration.urls.py`
+
+```python
+url(r"^controlmatrix/", include("integrations.controlmatrix.urls")),
+```
+
+## Testing with Mock Service
+
+The Example integration includes a mock Example service you can launch in the terminal to test your integration.
+
+To launch the mock service do the following in a separate terminal from the root directory of GovReady-Q:
+
+```python
+pip install click
+python integrations/controlmatrix/mock.py
+```

integrations/controlmatrix/communicate.py

@@ -0,0 +1,125 @@
+import requests
+import json
+from base64 import b64encode
+from urllib.parse import urlparse
+from integrations.utils.integration import Communication
+from integrations.models import Integration, Endpoint
+import pandas
+import pathlib
+
+
+class ControlmatrixCommunication(Communication):
+
+    DESCRIPTION = {
+        "name": "Controlmatrix",
+        "description": "Controlmatrix Service",
+        "version": "0.1",
+        "integration_db_record": False,
+        "mock": {
+            "base_url": "http:/localhost:9009",
+            "personal_access_token": None
+        }
+    }
+
+    def __init__(self, **kwargs):
+        assert self.DESCRIPTION, "Developer must assign a description dict"
+        self.__is_authenticated = False
+        self.error_msg = {}
+        self.auth_dict = {}
+        self.data = None
+        self.base_url = "https://controlmatrix.com/api"
+
+    def identify(self):
+        """Identify which Communication subclass"""
+        identity_str = f"This is {self.DESCRIPTION['name']} version {self.DESCRIPTION['version']}"
+        print(identity_str)
+        return identity_str
+
+    def setup(self, **kwargs):
+        pass
+
+    def get_response(self, endpoint, headers=None, verify=False):
+        response = requests.get(f"{self.base_url}{endpoint}")
+        self.status_code = response.status_code
+        if self.status_code == 200:
+            self.data = response.json()
+        elif self.status_code == 404:
+            print("404 - page not found")
+        else:
+            pass
+        return self.data
+
+    def authenticate(self, user=None, passwd=None):
+        """Authenticate with service"""
+        pass
+
+    @property
+    def is_authenticated(self):
+        return self.__is_authenticated
+
+    @is_authenticated.setter
+    def is_authenticated(self, value):
+        self.__is_authenticated = value
+
+    def extract_data(self, authentication, identifiers):
+        """Extract data"""
+        try:
+            data = []
+            rows_list = []
+            # TODO: Fix file path
+            fn = "integrations/controlmatrix/assets/data/controls_matrix.xlsx"
+            if pathlib.Path(fn).is_file():
+                try:
+                    df_dict = pandas.read_excel(fn, 'controls_foundation', header=0)
+                    for index, row in df_dict.iterrows():
+                        row_dict = {
+                            "CONTROL_STATUS": row.get('CONTROL_STATUS', ""),
+                            "MONITORED_STATUS": row.get('MONITORED_STATUS', ""),
+                            "TIER": row.get('TIER', ""),
+                            "REVIEW_FREQUENCY": row.get('REVIEW_FREQUENCY', ""),
+                            "CONTROL_TYPE": row.get('CONTROL_TYPE', ""),
+                            "AUTOMATED": row.get('AUTOMATED', ""),
+                            "COMMON_CONTROL": row.get('COMMON_CONTROL', ""),
+                            "PRIORITY": row.get('PRIORITY', ""),
+                            "BASELINE_IMPACT": row.get('BASELINE_IMPACT', ""),
+                            "RMF_CONTROL_FAMILY": row.get('RMF_CONTROL_FAMILY', ""),
+                            "RMF_CONTROL": row.get('RMF_CONTROL', ""),
+                            "RMF_CONTROL_NAME": row.get('RMF_CONTROL_NAME', ""),
+                            "CSF_FUNCTION": row.get('CSF_FUNCTION', ""),
+                            "CSF_CATEGORY": row.get('CSF_CATEGORY', ""),
+                            "CSF_SUBCATEGORY": row.get('CSF_SUBCATEGORY', ""),
+                            "CYBERSECURITY_PROGRAM_DOMAIN": row.get('CYBERSECURITY_PROGRAM_DOMAIN', ""),
+                            "ORTB": row.get('ORTB', ""),
+                            "FSA": row.get('FSA', ""),
+                            "HVA": row.get('HVA', ""),
+                            "FPKI": row.get('FPKI', ""),
+                            "PRIVACY": row.get('PRIVACY', ""),
+                            "CLOUD_SHARED": row.get('CLOUD_SHARED', ""),
+                            "FAST_TRACK": row.get('FAST_TRACK', "")
+                        }
+                        rows_list.append(row_dict)
+                except FileNotFoundError as e:
+                    print(f"Error reading file {fn}: {e}")
+                    # logger.error(f"Error reading file {fn}: {e}")
+                except Exception as e:
+                    # logger.error(f"Other error reading file {fn}: {e}")
+                    print(f"Other error reading file {fn}: {e}")
+            else:
+                print(f"Failed to find or open file '{fn}'.")
+
+            print(f"identifiers: ", identifiers)
+            # for ctl in rows_list:
+            #     print(ctl['RMF_CONTROL'].lower())
+            #     if ctl['RMF_CONTROL'].lower() in identifiers:
+            #         data.append(ctl)
+            control_matrix = next((ctl for ctl in rows_list if ctl['RMF_CONTROL'].lower() in identifiers), None)
+            data.append(control_matrix)
+        except:
+            data = []
+        return data
+
+    def transform_data(self, data, system_id=None, title=None, description=None, deployment_uuid=None):
+        pass
+
+    def load_data(self, data):
+        pass