Merge pull request #33 from CompassSecurity/improvements

Improvements

Author: fxai

.gitignore

@@ -135,8 +135,9 @@ dmypy.json
 *.key
 files/
 custom/testcases
+custom/testcaseskb
 
 supervisord.pid
 sampledata/sigma/*
 external/*
-INITIAL_ADMIN_PASSWORD.TXT
+INITIAL_ADMIN_PASSWORD.TXT

Dockerfile

@@ -1,5 +1,5 @@
 # pull official base image
-FROM python:3.11.3-slim-buster
+FROM python:3.11.13-slim
 
 # set work directory
 WORKDIR /usr/src/app
@@ -9,7 +9,7 @@ ENV PYTHONDONTWRITEBYTECODE 1
 ENV PYTHONUNBUFFERED 1
 
 # install system dependencies
-RUN apt-get update && apt-get install -y netcat git
+RUN apt-get update && apt-get install -y netcat-openbsd git
 
 # install dependencies
 RUN pip install --upgrade pip

README.md

@@ -13,14 +13,6 @@
   <a href="https://docs.purpleops.app"><img src="https://img.shields.io/badge/Docs-blue?logo=readthedocs&logoColor=white">
 </p>
 
-<p align="center">
-  <a href="#key-features">Key Features</a> •
-  <a href="#installation">Installation</a> •
-  <a href="#contact-us">Contact Us</a> •
-  <a href="#credits">Credit</a> •
-  <a href="#license">License</a>
-</p>
-
 <p align="center">
   <img src="static/images/demo.gif">
 </p>
@@ -133,15 +125,80 @@ $ sudo docker compose up
   ```
 </details>
 
-## Contact Us
+## Compass Security Fork
+The Compass Security fork includes fixes and new features!
 
-We would love to hear back from you, if something is broken or have and idea to make it better add a ticket or connect to us on the [PurpleOps Discord](https://discord.gg/2xeA6FB3GJ) or email us at pops@purpleops.app | `@_w_m__` 
+### Updated Dependencies
+The Python dependencies (e.g. Flask) were updated to the latest versions.
 
-## Credits
+### Restructured Test Case Form and Flow-Based Approach
+We have redesigned the test case form to prioritise the elements that we believe are important during a purple team engagement. 
+<br> Is there anything missing? Please let us know — we are eager to hear how other analysts approach Purple Teaming engagements.
+
+<br>Moreover, we have implemented a flow-based approach to facilitate collaboration with the Blue Team.
+<img width="2061" height="612" alt="image" src="https://github.com/user-attachments/assets/60e1d78d-cb73-4c10-ae16-9b8f296e59a1" />
+
+#### Waiting Blue:
+This signals to the blue team that input is expected from their side. Once the required information has been added, the Blue team can set the state to 'Waiting Red'.
+Users with the 'Blue' role can only edit a test case if it is in the 'Waiting Blue' or 'Waiting Red' state.
+
+#### Waiting Red:
+This signals to the red team that the blue team has finished adding their details to the test case. The red team can then check that all the required information is present. If so, the state can be changed to 'Complete'.
+
+#### Complete:
+The blue team cannot make any more changes to the test case.
+
+
+### Pytests
+We have created pytests for each route. This makes it easy to check whether the application has been affected by any changes made to it.
+<br><br>Note: We are still missing security checks (e.g. RBAC) and application logic checks, so if you would like to contribute, we would be glad to merge your pull request!
+
+### Dark Mode
+Enjoy PurpleOps in dark mode. To enable this, go to the settings menu.
+<img width="2063" height="1041" alt="darkmode_overview" src="https://github.com/user-attachments/assets/1b2870a6-319a-4ca6-8366-f5a8e638842e" />
+<img width="2066" height="1120" alt="darkmode_testcase" src="https://github.com/user-attachments/assets/056c721a-2a00-473f-8cbe-39537a843491" />
+
+### Test Case History
+The Test Case History allows you to view previous saved versions of the test case. This feature is only available after an initial save, not after an import. Please note that evidence files are not stored.
+![test_case_history](https://github.com/user-attachments/assets/b15c3799-a73b-4845-9ec3-fe9f04f3b7a4)
+
+### Restore Deleted Test Cases
+You can now restore deleted test cases (requires page reload).
+![test_case_restore](https://github.com/user-attachments/assets/a66ba8b0-854f-436b-9602-9c2a244e3ded)
+
+### Test Case Knowlege Base and Variables File
+We added the option to add an knowledge base MD file for each TPP. You can find an example here:
+https://github.com/CompassSecurity/PurpleOps/blob/main/custom/testcaseskb/T1087_002.md
 
+To view the KB click on the "compass" icon in the test case:
+![test_case_kb](https://github.com/user-attachments/assets/f06ed191-2812-4259-9317-ed6d865a729e)
+
+The KB also enables you to set placeholders for frequently used strings. For instance, you could define {{TARGET_DOMAIN_USER}} as a placeholder in an MD file for a command.
+```
+net user {{TARGET_DOMAIN_USER}} /domain 
+```
+Define a JSON file which contains all your placeholders and the coresponding text:
+```
+{
+"DOMAIN_NAME" : "testlab.local",
+"LOWPRIVILEGED_DOMAIN_USER" : "tmassie",
+"TARGET_DOMAIN_USER" : "administrator",
+"DC_IP" : "10.0.1.10"
+}
+```
+Upload the JSON file to PurpleOps using your browser. The values will be stored in your session storage (cleared after browser is closed). Use the toggle in the test case KB to replace the placeholders with real data.
+![test_case_kb_variables](https://github.com/user-attachments/assets/547f4a21-a043-47a9-8be4-c9a176a23029)
+
+
+## Credits
+- PurpleOps https://github.com/CyberCX-STA/PurpleOps
 - Atomic Red Team [(LICENSE)](https://github.com/redcanaryco/atomic-red-team/blob/master/LICENSE.txt) for sample commands
 - [CyberCX](https://cybercx.com.au/) for foundational support
 
 ## License
-
 Apache
+
+
+
+
+

blueprints/access.py

@@ -27,9 +27,18 @@ def users():
 @auth_required()
 @roles_accepted('Admin')
 def createuser():
+
+    email = request.form['email']
+    username = request.form['username']
+
+    if user_datastore.find_user(email=email):
+        return jsonify({'error': 'Email already exists'}), 400
+    if user_datastore.find_user(username=username):
+        return jsonify({'error': 'Username already exists'}), 400
+
     user = user_datastore.create_user(
-        email = request.form['email'],
-        username = request.form['username'],
+        email = email,
+        username = username,
         password = utils.hash_password(request.form['password']),
         roles = [Role.objects(name=role).first() for role in request.form.getlist('roles')],
         assessments = [Assessment.objects(name=assessment).first() for assessment in request.form.getlist('assessments')]
@@ -40,37 +49,55 @@ def createuser():
 @auth_required()
 @roles_accepted('Admin')
 def edituser(id):
-    origUser = User.objects(id=id).first()
     user = User.objects(id=id).first()
+    if not user:
+        return jsonify({'error': 'User not found'}), 404
+
     if request.method == 'POST':
+        new_email = request.form.get('email')
+        new_username = request.form.get('username')
+
+        # Check for email conflict
+        if new_email and User.objects(email=new_email, id__ne=id).first():
+            return jsonify({'error': 'Email already in use'}), 400
+
+        # Check for username conflict
+        if new_username and User.objects(username=new_username, id__ne=id).first():
+            return jsonify({'error': 'Username already in use'}), 400
+
         if "password" in request.form and request.form['password'].strip():
             user.password = utils.hash_password(request.form['password'])
 
         user = applyFormData(user, request.form, ["username", "email"])
-        # You cannot rename the inbuilt admin account
-        if origUser.username == "admin" and user.username != "admin":
+
+        # Prevent renaming built-in admin
+        if user.username != "admin" and User.objects(id=id).first().username == "admin":
             user.username = "admin"
 
-        user.roles = []
-        for role in request.form.getlist('roles'):
-            user.roles.append(Role.objects(name=role).first())
-        # You cannot de-admin the inbuilt admin, re-add admin wiped admin role
-        if user.username == "admin" and "Admin" not in [u.name for u in user.roles]:
+        # Update roles
+        user.roles = [
+            Role.objects(name=role).first()
+            for role in request.form.getlist('roles')
+        ]
+
+        # Ensure admin keeps the Admin role
+        if user.username == "admin" and "Admin" not in [r.name for r in user.roles]:
             user.roles.append(Role.objects(name="Admin").first())
 
-        user.assessments = []
-        for assessment in request.form.getlist('assessments'):
-            user.assessments.append(Assessment.objects(name=assessment).first())
-        # Admin users have implied access to all assessments, wipe selected assessments
-        if "Admin" in [u.name for u in user.roles]:
+        # Update assessments
+        user.assessments = [
+            Assessment.objects(name=assessment).first()
+            for assessment in request.form.getlist('assessments')
+        ]
+
+        # Admins get implicit access to all assessments
+        if "Admin" in [r.name for r in user.roles]:
             user.assessments = []
 
         user.save()
         return jsonify(user.to_json()), 200
-        
+
     if request.method == 'DELETE':
-        # Prevent inbuilt admin deletion
         if user.username != "admin":
             user.delete()
-            user.save()
         return "", 200

blueprints/assessment.py

@@ -51,7 +51,8 @@ def deleteassessment(id):
 def loadassessment(id):
     return render_template(
         'assessment.html',
-        testcases = TestCase.objects(assessmentid=id).all(),
+        testcases = TestCase.objects(assessmentid=id, deleted=False).all(),
+        deleted_testcases = TestCase.objects(assessmentid=id, deleted=True).all(),
         assessment = Assessment.objects(id=id).first(),
         templates = TestCaseTemplate.objects(),
         mitres = sorted(

blueprints/assessment_export.py

@@ -17,13 +17,13 @@
 @user_assigned_assessment
 def exportassessment(id, filetype):
     if filetype not in ["json", 'csv']:
-        return 401
+        return "", 401
 
     assessment = Assessment.objects(id=id).first()
     if current_user.has_role("Blue"):
-        testcases = TestCase.objects(assessmentid=str(assessment.id), visible=True).all()
+        testcases = TestCase.objects(assessmentid=str(assessment.id), visible=True, deleted=False).all()
     else:
-        testcases = TestCase.objects(assessmentid=str(assessment.id)).all()
+        testcases = TestCase.objects(assessmentid=str(assessment.id), deleted=False).all()
 
     jsonDict = []
     for testcase in testcases:
@@ -57,9 +57,9 @@ def exportassessment(id, filetype):
 def exportcampaign(id):
     assessment = Assessment.objects(id=id).first()
     if current_user.has_role("Blue"):
-        testcases = TestCase.objects(assessmentid=str(assessment.id), visible=True).all()
+        testcases = TestCase.objects(assessmentid=str(assessment.id), visible=True, deleted=False).all()
     else:
-        testcases = TestCase.objects(assessmentid=str(assessment.id)).all()
+        testcases = TestCase.objects(assessmentid=str(assessment.id), deleted=False).all()
 
     jsonDict = []
     for testcase in testcases:
@@ -156,9 +156,9 @@ def exportnavigator(id):
     }
 
     if current_user.has_role("Blue"):
-        testcases = TestCase.objects(assessmentid=id, visible=True).all()
+        testcases = TestCase.objects(assessmentid=id, visible=True, deleted=False).all()
     else:
-        testcases = TestCase.objects(assessmentid=id).all()
+        testcases = TestCase.objects(assessmentid=id, deleted=False).all()
 
     results = {}
     for testcase in testcases:
@@ -214,7 +214,7 @@ def exportentire(id):
     else:
         # If they're blue then they can only export the evidence files of visible testcases
         shutil.copytree(f"files/{id}", f"files/tmp{id}")
-        testcases = TestCase.objects(assessmentid=str(assessment.id)).all()
+        testcases = TestCase.objects(assessmentid=str(assessment.id), deleted=False).all()
         for testcase in testcases:
             if not testcase.visible and os.path.isdir(f"files/tmp{id}/{str(testcase.id)}"):
                 shutil.rmtree(f"files/tmp{id}/{str(testcase.id)}")

blueprints/assessment_utils.py

@@ -56,8 +56,7 @@ def assessmentmulti(id, field):
 def assessmentnavigator(id):
     assessment = Assessment.objects(id=id).first()
 
-    # Create and store one-time secret; timestamp and ip for later comparison in
-    # the unauthed `thisurl`.json endpoint
+    # Create and store one-time secret; timestamp and ip for later comparison in the unauthed navigator.json endpoint
     secret = secrets.token_urlsafe()
     assessment.navigatorexport = f"{int(time())}|{request.remote_addr}|{secret}"
     assessment.save()
@@ -69,22 +68,33 @@ def assessmentnavigator(id):
 @blueprint_assessment_utils.route('/assessment/<id>/navigator.json', methods = ['GET'])
 def assessmentnavigatorjson(id):
     assessment = Assessment.objects(id=id).first()
-    timestamp, ip, secret = assessment.navigatorexport.split("|")
-    
-    # This endpoint is unauthed so that we can embed the ATT&CK Navigator and
-    # allow it to fetch a layer.json on behalf of the user. To mitigate security issues
-    # the endpoint needs to be hit
-    # 1. Within 10 seconds of hitting the authed endpoint /assessment/<id>/navigator
-    # 2. With the same IP used to his the above authed endpoint
-    # 3. With a one-time secret key returned in the above authed endpoint
-    # 4. From the mitre-attack origin (yes this is spoofable, but why not)
-    # if (int(time()) - int(timestamp) <= 30 and
-        #request.remote_addr == ip and
-        # request.args.get("secret") == secret): # and
-        #request.origin == "https://mitre-attack.github.io"):
-    response = make_response(send_from_directory('files', f"{id}/navigator.json"))
-    response.headers.add('Access-Control-Allow-Origin', '*')
-    return response
+    if not assessment or not assessment.navigatorexport:
+        return "", 401
+
+    try:
+        timestamp, ip, secret = assessment.navigatorexport.split("|")
+    except ValueError:
+        return "", 401
+
+    request_ip = request.remote_addr
+    request_secret = request.args.get("secret", type=str)
+    request_origin = request.headers.get("Origin")
+
+    # Validate conditions
+    if (
+        int(time()) - int(timestamp) <= 10 and
+        request_ip == ip and
+        request_secret == secret and
+        request_origin == "https://mitre-attack.github.io"
+    ):
+        response = make_response(send_from_directory('files', f"{id}/navigator.json"))
+        response.headers.add('Access-Control-Allow-Origin', '*')
+        
+        # Clear the one-time secret to prevent reuse
+        assessment.navigatorexport = None
+        assessment.save()
+
+        return response
 
     return "", 401
 
@@ -94,9 +104,9 @@ def assessmentnavigatorjson(id):
 def assessmentstats(id):
     assessment = Assessment.objects(id=id).first()
     if current_user.has_role("Blue"):
-        testcases = TestCase.objects(assessmentid=str(assessment.id), visible=True).all()
+        testcases = TestCase.objects(assessmentid=str(assessment.id), visible=True, deleted=False).all()
     else:
-        testcases = TestCase.objects(assessmentid=str(assessment.id)).all()
+        testcases = TestCase.objects(assessmentid=str(assessment.id), deleted=False).all()
 
     # Initalise metrics that are captured
     stats = {
@@ -169,20 +179,20 @@ def assessmenthexagons(id):
     shownHexs = []
     hiddenHexs = []
     for i in range(len(tactics)):
-        if not TestCase.objects(assessmentid=id, tactic=tactics[i], state="Complete").count():
+        if not TestCase.objects(assessmentid=id, tactic=tactics[i], state="Complete", deleted=False).count():
             hiddenHexs.append({
                 "display": "none",
-                "stroke": "#ffffff",
-                "fill": "#ffffff",
-                "arrow": "rgba(0, 0, 0, 0)",
+                "stroke": "none",
+                "fill": "none",
+                "arrow": "none",
                 "text": ""
             })
             continue
 
         cumulatedscore = 0
         count = 0
 
-        for testcase in TestCase.objects(assessmentid=id, tactic=tactics[i]):
+        for testcase in TestCase.objects(assessmentid=id, tactic=tactics[i], deleted=False):
             if testcase.testcasescore is not None:
                 cumulatedscore += testcase.testcasescore
                 count += 1
@@ -201,15 +211,15 @@ def assessmenthexagons(id):
                 "display": "block",
                 "stroke": color,
                 "fill": "#eeeeee",
-                "arrow": "rgb(0, 0, 0)",
+                "arrow": "#593196",
                 "text": tactics[i]
             })
         else:
             hiddenHexs.append({
                 "display": "none",
-                "stroke": "#ffffff",
-                "fill": "#ffffff",
-                "arrow": "rgba(0, 0, 0, 0)",
+                "stroke": "none",
+                "fill": "none",
+                "arrow": "none",
                 "text": ""
             })