From 08250990e3fa81baea6d8ad6767efda78135c2d4 Mon Sep 17 00:00:00 2001
From: deacon <marksoccerman1@aol.com>
Date: Mon, 16 Mar 2026 10:37:29 -0400
Subject: [PATCH 01/10] feat: add YAML ability file upload endpoint (POST
 /api/v2/abilities/upload)

---
 app/api/v2/handlers/ability_api.py           |  19 +++
 app/api/v2/managers/ability_api_manager.py   |  70 +++++++++++
 tests/api/v2/handlers/test_ability_upload.py | 120 +++++++++++++++++++
 3 files changed, 209 insertions(+)
 create mode 100644 tests/api/v2/handlers/test_ability_upload.py

diff --git a/app/api/v2/handlers/ability_api.py b/app/api/v2/handlers/ability_api.py
index 2aaa14c37..88ca3c329 100644
--- a/app/api/v2/handlers/ability_api.py
+++ b/app/api/v2/handlers/ability_api.py
@@ -16,6 +16,7 @@ def __init__(self, services):
     def add_routes(self, app: web.Application):
         router = app.router
         router.add_get('/abilities', self.get_abilities)
+        router.add_post('/abilities/upload', self.upload_ability)
         router.add_get('/abilities/{ability_id}', self.get_ability_by_id)
         router.add_post('/abilities', self.create_ability)
         router.add_put('/abilities/{ability_id}', self.create_or_update_ability)
@@ -110,3 +111,21 @@ async def update_ability(self, request: web.Request):
     async def delete_ability(self, request: web.Request):
         await self.delete_on_disk_object(request)
         return web.HTTPNoContent()
+
+    @aiohttp_apispec.docs(tags=['abilities'],
+                          summary='Upload a YAML ability file.',
+                          description='Uploads a YAML ability file, validates its contents, '
+                                      'saves it to disk under data/abilities/{tactic}/, '
+                                      'and loads it into memory.')
+    @aiohttp_apispec.response_schema(AbilitySchema,
+                                     description='JSON dictionary representation of the uploaded Ability.')
+    async def upload_ability(self, request: web.Request):
+        reader = await request.multipart()
+        file_field = await reader.next()
+        if not file_field or file_field.name != 'file':
+            raise web.HTTPBadRequest(reason='Missing "file" field in multipart form data.')
+        filename = file_field.filename or ''
+        file_data = await file_field.read()
+        access = await self.get_request_permissions(request)
+        ability = await self._api_manager.upload_ability_file(file_data, filename, access)
+        return web.json_response(ability.display)
diff --git a/app/api/v2/managers/ability_api_manager.py b/app/api/v2/managers/ability_api_manager.py
index 8dc6be270..00d6713f4 100644
--- a/app/api/v2/managers/ability_api_manager.py
+++ b/app/api/v2/managers/ability_api_manager.py
@@ -90,6 +90,76 @@ def _create_ability_filepath(self, tactic: str, obj_id: str):
             os.makedirs(tactic_dir)
         return os.path.join(tactic_dir, '%s.yml' % obj_id)
 
+    async def upload_ability_file(self, file_data: bytes, filename: str, access: dict):
+        """Upload a YAML ability file, validate it, save to disk, and load into memory."""
+        # Validate file extension
+        if not filename.lower().endswith(('.yml', '.yaml')):
+            raise JsonHttpBadRequest('Invalid file type. Only .yml and .yaml files are accepted.')
+
+        # Parse YAML
+        try:
+            parsed = yaml.safe_load(file_data)
+        except yaml.YAMLError as e:
+            raise JsonHttpBadRequest(f'Invalid YAML: {e}')
+
+        # Handle list-wrapped abilities (common format: list of one dict)
+        if isinstance(parsed, list):
+            if len(parsed) == 0:
+                raise JsonHttpBadRequest('YAML file contains an empty list.')
+            parsed = parsed[0]
+
+        if not isinstance(parsed, dict):
+            raise JsonHttpBadRequest('YAML file must contain a mapping/dictionary.')
+
+        # Extract required fields
+        ability_id = parsed.get('id') or parsed.get('ability_id')
+        if not ability_id:
+            raise JsonHttpBadRequest('Missing required field: "id" or "ability_id".')
+        name = parsed.get('name')
+        if not name:
+            raise JsonHttpBadRequest('Missing required field: "name".')
+        tactic = parsed.get('tactic')
+        if not tactic:
+            raise JsonHttpBadRequest('Missing required field: "tactic".')
+
+        # Validate ID and tactic format
+        validator = re.compile(r'^[a-zA-Z0-9-_]+$')
+        if not validator.match(str(ability_id)):
+            raise JsonHttpBadRequest(f'Invalid ability ID "{ability_id}". '
+                                     'IDs can only contain alphanumeric characters, hyphens, and underscores.')
+        if not validator.match(tactic):
+            raise JsonHttpBadRequest(f'Invalid tactic "{tactic}". '
+                                     'Tactics can only contain alphanumeric characters, hyphens, and underscores.')
+
+        tactic = tactic.lower()
+        parsed['tactic'] = tactic
+        parsed['id'] = ability_id
+
+        # Check for duplicates
+        existing = list(self.find_objects('abilities', dict(ability_id=str(ability_id))))
+        if existing:
+            raise JsonHttpBadRequest(f'Ability with id already exists: {ability_id}')
+
+        # Determine save path and create directory if needed
+        tactic_dir = os.path.join('data', 'abilities', tactic)
+        if not os.path.exists(tactic_dir):
+            os.makedirs(tactic_dir)
+        file_path = os.path.join(tactic_dir, f'{ability_id}.yml')
+
+        # Write the file
+        with open(file_path, 'wb') as f:
+            f.write(yaml.dump([parsed], encoding='utf-8', sort_keys=False))
+
+        # Load into memory
+        allowed = self._get_allowed_from_access(access)
+        await self._data_svc.load_ability_file(file_path, allowed)
+
+        # Return the loaded ability
+        loaded = self.find_objects('abilities', dict(ability_id=str(ability_id)))
+        if loaded:
+            return list(loaded)[0]
+        raise JsonHttpBadRequest(f'Ability was saved but could not be loaded into memory: {ability_id}')
+
     async def _save_and_reload_object(self, file_path: str, data: dict, obj_type: type, access: BaseWorld.Access):
         await self._file_svc.save_file(file_path, yaml.dump([data], encoding='utf-8', sort_keys=False),
                                        '', encrypt=False)
diff --git a/tests/api/v2/handlers/test_ability_upload.py b/tests/api/v2/handlers/test_ability_upload.py
new file mode 100644
index 000000000..df574bb0c
--- /dev/null
+++ b/tests/api/v2/handlers/test_ability_upload.py
@@ -0,0 +1,120 @@
+import os
+import pytest
+import yaml
+
+from aiohttp import FormData
+from http import HTTPStatus
+
+
+@pytest.fixture
+def valid_ability_yaml():
+    ability_data = {
+        'id': 'upload-test-001',
+        'name': 'Uploaded Test Ability',
+        'description': 'An ability uploaded via YAML file',
+        'tactic': 'discovery',
+        'technique_id': 'T1082',
+        'technique_name': 'System Information Discovery',
+        'executors': [
+            {
+                'name': 'sh',
+                'platform': 'linux',
+                'command': 'uname -a'
+            }
+        ]
+    }
+    yield yaml.dump([ability_data], sort_keys=False).encode('utf-8')
+
+    # Cleanup
+    path = 'data/abilities/discovery/upload-test-001.yml'
+    if os.path.exists(path):
+        try:
+            os.remove(path)
+        except OSError:
+            pass
+
+
+@pytest.fixture
+def valid_ability_yaml_with_ability_id_key():
+    """Uses 'ability_id' key instead of 'id'."""
+    ability_data = {
+        'ability_id': 'upload-test-002',
+        'name': 'Uploaded Test Ability 2',
+        'description': 'An ability using ability_id key',
+        'tactic': 'collection',
+        'technique_id': 'T1005',
+        'technique_name': 'Data from Local System',
+        'executors': [
+            {
+                'name': 'sh',
+                'platform': 'linux',
+                'command': 'ls -la'
+            }
+        ]
+    }
+    yield yaml.dump([ability_data], sort_keys=False).encode('utf-8')
+
+    path = 'data/abilities/collection/upload-test-002.yml'
+    if os.path.exists(path):
+        try:
+            os.remove(path)
+        except OSError:
+            pass
+
+
+class TestAbilityUploadApi:
+
+    async def test_upload_valid_yaml(self, api_v2_client, api_cookies, valid_ability_yaml):
+        form = FormData()
+        form.add_field('file', valid_ability_yaml,
+                       filename='upload-test-001.yml',
+                       content_type='application/x-yaml')
+        resp = await api_v2_client.post('/api/v2/abilities/upload',
+                                        cookies=api_cookies,
+                                        data=form)
+        assert resp.status == HTTPStatus.OK
+        result = await resp.json()
+        assert result['ability_id'] == 'upload-test-001'
+        assert result['name'] == 'Uploaded Test Ability'
+        assert result['tactic'] == 'discovery'
+        assert os.path.exists('data/abilities/discovery/upload-test-001.yml')
+
+    async def test_upload_invalid_file_type(self, api_v2_client, api_cookies):
+        form = FormData()
+        form.add_field('file', b'not yaml content',
+                       filename='bad_file.txt',
+                       content_type='text/plain')
+        resp = await api_v2_client.post('/api/v2/abilities/upload',
+                                        cookies=api_cookies,
+                                        data=form)
+        assert resp.status == HTTPStatus.BAD_REQUEST
+
+    async def test_upload_malformed_yaml(self, api_v2_client, api_cookies):
+        form = FormData()
+        form.add_field('file', b'{{invalid: yaml: [}',
+                       filename='malformed.yml',
+                       content_type='application/x-yaml')
+        resp = await api_v2_client.post('/api/v2/abilities/upload',
+                                        cookies=api_cookies,
+                                        data=form)
+        assert resp.status == HTTPStatus.BAD_REQUEST
+
+    async def test_upload_missing_required_fields(self, api_v2_client, api_cookies):
+        incomplete = yaml.dump([{'description': 'no id, name, or tactic'}]).encode('utf-8')
+        form = FormData()
+        form.add_field('file', incomplete,
+                       filename='incomplete.yml',
+                       content_type='application/x-yaml')
+        resp = await api_v2_client.post('/api/v2/abilities/upload',
+                                        cookies=api_cookies,
+                                        data=form)
+        assert resp.status == HTTPStatus.BAD_REQUEST
+
+    async def test_unauthorized_upload(self, api_v2_client, valid_ability_yaml):
+        form = FormData()
+        form.add_field('file', valid_ability_yaml,
+                       filename='upload-test-001.yml',
+                       content_type='application/x-yaml')
+        resp = await api_v2_client.post('/api/v2/abilities/upload',
+                                        data=form)
+        assert resp.status == HTTPStatus.UNAUTHORIZED

From cc5b097b6a91052b996d1d908ba0923f9bae5fbb Mon Sep 17 00:00:00 2001
From: Fiona McCrae <35509590+fionamccrae@users.noreply.github.com>
Date: Thu, 23 Apr 2026 11:24:48 -0400
Subject: [PATCH 02/10] Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
---
 app/api/v2/handlers/ability_api.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/app/api/v2/handlers/ability_api.py b/app/api/v2/handlers/ability_api.py
index 88ca3c329..cc7d5f49d 100644
--- a/app/api/v2/handlers/ability_api.py
+++ b/app/api/v2/handlers/ability_api.py
@@ -1,5 +1,6 @@
 import aiohttp_apispec
 from aiohttp import web
+import json
 
 from app.api.v2.handlers.base_object_api import BaseObjectApi
 from app.api.v2.managers.ability_api_manager import AbilityApiManager
@@ -123,7 +124,8 @@ async def upload_ability(self, request: web.Request):
         reader = await request.multipart()
         file_field = await reader.next()
         if not file_field or file_field.name != 'file':
-            raise web.HTTPBadRequest(reason='Missing "file" field in multipart form data.')
+            error_body = json.dumps({'error': 'Missing "file" field in multipart form data.'})
+            raise web.HTTPBadRequest(text=error_body, content_type='application/json')
         filename = file_field.filename or ''
         file_data = await file_field.read()
         access = await self.get_request_permissions(request)

From 4b31fafaa05c2adb05058752a72316fd87960626 Mon Sep 17 00:00:00 2001
From: Fiona McCrae <fmccrae@mitre.org>
Date: Thu, 23 Apr 2026 13:38:40 -0400
Subject: [PATCH 03/10] added test for ability_id format

---
 tests/api/v2/handlers/test_ability_upload.py | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/tests/api/v2/handlers/test_ability_upload.py b/tests/api/v2/handlers/test_ability_upload.py
index df574bb0c..113b1974f 100644
--- a/tests/api/v2/handlers/test_ability_upload.py
+++ b/tests/api/v2/handlers/test_ability_upload.py
@@ -79,6 +79,21 @@ async def test_upload_valid_yaml(self, api_v2_client, api_cookies, valid_ability
         assert result['tactic'] == 'discovery'
         assert os.path.exists('data/abilities/discovery/upload-test-001.yml')
 
+    async def test_upload_valid_yaml_ability_id(self, api_v2_client, api_cookies, valid_ability_yaml_with_ability_id_key):
+        form = FormData()
+        form.add_field('file', valid_ability_yaml_with_ability_id_key,
+                       filename='upload-test-001.yml',
+                       content_type='application/x-yaml')
+        resp = await api_v2_client.post('/api/v2/abilities/upload',
+                                        cookies=api_cookies,
+                                        data=form)
+        assert resp.status == HTTPStatus.OK
+        result = await resp.json()
+        assert result['ability_id'] == 'upload-test-002'
+        assert result['name'] == 'Uploaded Test Ability 2'
+        assert result['tactic'] == 'collection'
+        assert os.path.exists('data/abilities/collection/upload-test-002.yml')
+
     async def test_upload_invalid_file_type(self, api_v2_client, api_cookies):
         form = FormData()
         form.add_field('file', b'not yaml content',

From 9a5b000258bfa415a77605bf355c2c16559847cc Mon Sep 17 00:00:00 2001
From: Fiona McCrae <fmccrae@mitre.org>
Date: Thu, 23 Apr 2026 16:21:27 -0400
Subject: [PATCH 04/10] tweaks allowing tests to pass by updating file
 processing, need to update swaggerdocs and confirm security:

---
 app/api/v2/handlers/ability_api.py         |  3 ++-
 app/api/v2/managers/ability_api_manager.py | 15 ++++++++++++++-
 2 files changed, 16 insertions(+), 2 deletions(-)

diff --git a/app/api/v2/handlers/ability_api.py b/app/api/v2/handlers/ability_api.py
index cc7d5f49d..eefc63a98 100644
--- a/app/api/v2/handlers/ability_api.py
+++ b/app/api/v2/handlers/ability_api.py
@@ -118,13 +118,14 @@ async def delete_ability(self, request: web.Request):
                           description='Uploads a YAML ability file, validates its contents, '
                                       'saves it to disk under data/abilities/{tactic}/, '
                                       'and loads it into memory.')
+    # add request_schema here
     @aiohttp_apispec.response_schema(AbilitySchema,
                                      description='JSON dictionary representation of the uploaded Ability.')
     async def upload_ability(self, request: web.Request):
         reader = await request.multipart()
         file_field = await reader.next()
         if not file_field or file_field.name != 'file':
-            error_body = json.dumps({'error': 'Missing "file" field in multipart form data.'})
+            error_body = json.dumps({'error': 'Missing "file" field as first in multipart form data.'})
             raise web.HTTPBadRequest(text=error_body, content_type='application/json')
         filename = file_field.filename or ''
         file_data = await file_field.read()
diff --git a/app/api/v2/managers/ability_api_manager.py b/app/api/v2/managers/ability_api_manager.py
index 00d6713f4..4ec94e97d 100644
--- a/app/api/v2/managers/ability_api_manager.py
+++ b/app/api/v2/managers/ability_api_manager.py
@@ -98,7 +98,17 @@ async def upload_ability_file(self, file_data: bytes, filename: str, access: dic
 
         # Parse YAML
         try:
-            parsed = yaml.safe_load(file_data)
+            # Ensure we pass a text string to yaml.safe_load. PyYAML accepts strings or file-like objects.
+            # The test fixtures pass bytes; decode bytes/bytearray to str before parsing.
+            if isinstance(file_data, (bytes, bytearray)):
+                try:
+                    file_text = file_data.decode('utf-8')
+                except UnicodeDecodeError:
+                    # Fallback to latin-1 to preserve byte values if utf-8 fails
+                    file_text = file_data.decode('latin-1')
+            else:
+                file_text = file_data
+            parsed = yaml.safe_load(file_text)
         except yaml.YAMLError as e:
             raise JsonHttpBadRequest(f'Invalid YAML: {e}')
 
@@ -134,6 +144,8 @@ async def upload_ability_file(self, file_data: bytes, filename: str, access: dic
         tactic = tactic.lower()
         parsed['tactic'] = tactic
         parsed['id'] = ability_id
+        # Normalize keys so downstream loaders don't receive duplicate ability_id
+        parsed.pop('ability_id', None)
 
         # Check for duplicates
         existing = list(self.find_objects('abilities', dict(ability_id=str(ability_id))))
@@ -148,6 +160,7 @@ async def upload_ability_file(self, file_data: bytes, filename: str, access: dic
 
         # Write the file
         with open(file_path, 'wb') as f:
+            # Dump as bytes using explicit encoding
             f.write(yaml.dump([parsed], encoding='utf-8', sort_keys=False))
 
         # Load into memory

From c353b7f3d234c30a6c44c09fd713f79dd5c5d9e1 Mon Sep 17 00:00:00 2001
From: Fiona McCrae <fmccrae@mitre.org>
Date: Thu, 23 Apr 2026 17:54:44 -0400
Subject: [PATCH 05/10] simplified upload to use pre-existing validation and
 writing code from JSON create ability API endpoint

---
 app/api/v2/handlers/ability_api.py         | 17 +++---
 app/api/v2/managers/ability_api_manager.py | 64 +++-------------------
 app/api/v2/schemas/ability_schemas.py      | 38 +++++++++++++
 3 files changed, 57 insertions(+), 62 deletions(-)
 create mode 100644 app/api/v2/schemas/ability_schemas.py

diff --git a/app/api/v2/handlers/ability_api.py b/app/api/v2/handlers/ability_api.py
index eefc63a98..d3f3493fd 100644
--- a/app/api/v2/handlers/ability_api.py
+++ b/app/api/v2/handlers/ability_api.py
@@ -6,6 +6,7 @@
 from app.api.v2.managers.ability_api_manager import AbilityApiManager
 from app.api.v2.schemas.base_schemas import BaseGetAllQuerySchema, BaseGetOneQuerySchema
 from app.objects.c_ability import Ability, AbilitySchema
+from app.api.v2.schemas.ability_schemas import AbilityUploadRequestSchema
 
 
 class AbilityApi(BaseObjectApi):
@@ -118,17 +119,19 @@ async def delete_ability(self, request: web.Request):
                           description='Uploads a YAML ability file, validates its contents, '
                                       'saves it to disk under data/abilities/{tactic}/, '
                                       'and loads it into memory.')
-    # add request_schema here
+    @aiohttp_apispec.form_schema(AbilityUploadRequestSchema)
     @aiohttp_apispec.response_schema(AbilitySchema,
                                      description='JSON dictionary representation of the uploaded Ability.')
     async def upload_ability(self, request: web.Request):
-        reader = await request.multipart()
-        file_field = await reader.next()
-        if not file_field or file_field.name != 'file':
-            error_body = json.dumps({'error': 'Missing "file" field as first in multipart form data.'})
-            raise web.HTTPBadRequest(text=error_body, content_type='application/json')
+        # aiohttp_apispec.form_schema() already parses multipart data and populates request["form"].
+        # Use the pre-parsed form to avoid re-reading the request body which causes multipart boundary errors.
+        file_field: web.FileField = request.get("form", {}).get("file")
+        if not file_field:
+            return json.dumps({'error': 'Missing "file" field in multipart form data.'})
         filename = file_field.filename or ''
-        file_data = await file_field.read()
+        # file_field.file is a synchronous file-like object; read it in a thread to avoid blocking the event loop.
+        loop = __import__('asyncio').get_event_loop()
+        file_data = await loop.run_in_executor(None, file_field.file.read)
         access = await self.get_request_permissions(request)
         ability = await self._api_manager.upload_ability_file(file_data, filename, access)
         return web.json_response(ability.display)
diff --git a/app/api/v2/managers/ability_api_manager.py b/app/api/v2/managers/ability_api_manager.py
index 4ec94e97d..f0b3f7dfa 100644
--- a/app/api/v2/managers/ability_api_manager.py
+++ b/app/api/v2/managers/ability_api_manager.py
@@ -7,7 +7,7 @@
 
 from app.api.v2.managers.base_api_manager import BaseApiManager
 from app.api.v2.responses import JsonHttpBadRequest
-from app.objects.c_ability import AbilitySchema
+from app.objects.c_ability import AbilitySchema, Ability
 from app.utility.base_world import BaseWorld
 
 
@@ -51,8 +51,11 @@ async def update_on_disk_object(self, obj: Any, data: dict, ram_key: str, id_pro
         return next(self.find_objects(ram_key, {id_property: obj_id}))
 
     def _validate_ability_data(self, create: bool, data: dict):
-        # Correct ability_id key for ability file saving.
-        data['id'] = data.pop('ability_id', '')
+        # Normalize ability ID: prefer explicit 'ability_id' if provided, otherwise preserve any existing 'id'.
+        if 'ability_id' in data:
+            data['id'] = data.pop('ability_id')
+        else:
+            data['id'] = data.get('id', '')
 
         # If a new ability is being created, ensure required fields present.
         if create:
@@ -99,7 +102,6 @@ async def upload_ability_file(self, file_data: bytes, filename: str, access: dic
         # Parse YAML
         try:
             # Ensure we pass a text string to yaml.safe_load. PyYAML accepts strings or file-like objects.
-            # The test fixtures pass bytes; decode bytes/bytearray to str before parsing.
             if isinstance(file_data, (bytes, bytearray)):
                 try:
                     file_text = file_data.decode('utf-8')
@@ -121,57 +123,9 @@ async def upload_ability_file(self, file_data: bytes, filename: str, access: dic
         if not isinstance(parsed, dict):
             raise JsonHttpBadRequest('YAML file must contain a mapping/dictionary.')
 
-        # Extract required fields
-        ability_id = parsed.get('id') or parsed.get('ability_id')
-        if not ability_id:
-            raise JsonHttpBadRequest('Missing required field: "id" or "ability_id".')
-        name = parsed.get('name')
-        if not name:
-            raise JsonHttpBadRequest('Missing required field: "name".')
-        tactic = parsed.get('tactic')
-        if not tactic:
-            raise JsonHttpBadRequest('Missing required field: "tactic".')
-
-        # Validate ID and tactic format
-        validator = re.compile(r'^[a-zA-Z0-9-_]+$')
-        if not validator.match(str(ability_id)):
-            raise JsonHttpBadRequest(f'Invalid ability ID "{ability_id}". '
-                                     'IDs can only contain alphanumeric characters, hyphens, and underscores.')
-        if not validator.match(tactic):
-            raise JsonHttpBadRequest(f'Invalid tactic "{tactic}". '
-                                     'Tactics can only contain alphanumeric characters, hyphens, and underscores.')
-
-        tactic = tactic.lower()
-        parsed['tactic'] = tactic
-        parsed['id'] = ability_id
-        # Normalize keys so downstream loaders don't receive duplicate ability_id
-        parsed.pop('ability_id', None)
-
-        # Check for duplicates
-        existing = list(self.find_objects('abilities', dict(ability_id=str(ability_id))))
-        if existing:
-            raise JsonHttpBadRequest(f'Ability with id already exists: {ability_id}')
-
-        # Determine save path and create directory if needed
-        tactic_dir = os.path.join('data', 'abilities', tactic)
-        if not os.path.exists(tactic_dir):
-            os.makedirs(tactic_dir)
-        file_path = os.path.join(tactic_dir, f'{ability_id}.yml')
-
-        # Write the file
-        with open(file_path, 'wb') as f:
-            # Dump as bytes using explicit encoding
-            f.write(yaml.dump([parsed], encoding='utf-8', sort_keys=False))
-
-        # Load into memory
-        allowed = self._get_allowed_from_access(access)
-        await self._data_svc.load_ability_file(file_path, allowed)
-
-        # Return the loaded ability
-        loaded = self.find_objects('abilities', dict(ability_id=str(ability_id)))
-        if loaded:
-            return list(loaded)[0]
-        raise JsonHttpBadRequest(f'Ability was saved but could not be loaded into memory: {ability_id}')
+        # write the file and return the created ability object
+        return await self.create_on_disk_object(data=parsed, access=access,
+                                                ram_key='abilities', id_property='ability_id', obj_class=Ability)
 
     async def _save_and_reload_object(self, file_path: str, data: dict, obj_type: type, access: BaseWorld.Access):
         await self._file_svc.save_file(file_path, yaml.dump([data], encoding='utf-8', sort_keys=False),
diff --git a/app/api/v2/schemas/ability_schemas.py b/app/api/v2/schemas/ability_schemas.py
new file mode 100644
index 000000000..0a4f53ce9
--- /dev/null
+++ b/app/api/v2/schemas/ability_schemas.py
@@ -0,0 +1,38 @@
+from marshmallow import fields, schema
+
+
+class AbilityUploadRequestSchema(schema.Schema):
+    """Schema for the multipart upload endpoint expecting a YAML file containing an ability.
+
+    The `file` field is a raw file upload. The metadata contains a short description and an
+    example YAML payload that matches the tests in `tests/api/v2/handlers/test_ability_upload.py`.
+    """
+
+    file = fields.Raw(required=True, metadata={
+        'type': 'file',
+        'description': 'A YAML file containing one ability (or a list with one ability). The tests send a YAML list with a single ability entry.',
+        'example': (
+            "- id: upload-test-001\n"
+            "  name: Uploaded Test Ability\n"
+            "  description: An ability uploaded via YAML file\n"
+            "  tactic: discovery\n"
+            "  technique_id: T1082\n"
+            "  technique_name: System Information Discovery\n"
+            "  executors:\n"
+            "  - name: sh\n"
+            "    platform: linux\n"
+            "    command: uname -a\n"
+        ),
+        'example_alternative': (
+            "- ability_id: upload-test-002\n"
+            "  name: Uploaded Test Ability 2\n"
+            "  description: An ability using ability_id key\n"
+            "  tactic: collection\n"
+            "  technique_id: T1005\n"
+            "  technique_name: Data from Local System\n"
+            "  executors:\n"
+            "  - name: sh\n"
+            "    platform: linux\n"
+            "    command: ls -la\n"
+        )
+    })

From f1efdcd909de9a73d099a882f56880591a5b2f69 Mon Sep 17 00:00:00 2001
From: Fiona McCrae <fmccrae@mitre.org>
Date: Thu, 30 Apr 2026 15:10:36 -0400
Subject: [PATCH 06/10] removing unnecessary upload api endpoint and modifying
 ability upload on caldera api to be more similar to adversary upload api

---
 app/api/v2/handlers/ability_api.py           |  25 ---
 app/api/v2/managers/ability_api_manager.py   |  48 +----
 app/api/v2/schemas/ability_schemas.py        |  38 ----
 app/objects/c_ability.py                     |  56 +++++-
 tests/api/v2/handlers/test_ability_upload.py | 180 +++++++++----------
 5 files changed, 147 insertions(+), 200 deletions(-)
 delete mode 100644 app/api/v2/schemas/ability_schemas.py

diff --git a/app/api/v2/handlers/ability_api.py b/app/api/v2/handlers/ability_api.py
index d3f3493fd..2aaa14c37 100644
--- a/app/api/v2/handlers/ability_api.py
+++ b/app/api/v2/handlers/ability_api.py
@@ -1,12 +1,10 @@
 import aiohttp_apispec
 from aiohttp import web
-import json
 
 from app.api.v2.handlers.base_object_api import BaseObjectApi
 from app.api.v2.managers.ability_api_manager import AbilityApiManager
 from app.api.v2.schemas.base_schemas import BaseGetAllQuerySchema, BaseGetOneQuerySchema
 from app.objects.c_ability import Ability, AbilitySchema
-from app.api.v2.schemas.ability_schemas import AbilityUploadRequestSchema
 
 
 class AbilityApi(BaseObjectApi):
@@ -18,7 +16,6 @@ def __init__(self, services):
     def add_routes(self, app: web.Application):
         router = app.router
         router.add_get('/abilities', self.get_abilities)
-        router.add_post('/abilities/upload', self.upload_ability)
         router.add_get('/abilities/{ability_id}', self.get_ability_by_id)
         router.add_post('/abilities', self.create_ability)
         router.add_put('/abilities/{ability_id}', self.create_or_update_ability)
@@ -113,25 +110,3 @@ async def update_ability(self, request: web.Request):
     async def delete_ability(self, request: web.Request):
         await self.delete_on_disk_object(request)
         return web.HTTPNoContent()
-
-    @aiohttp_apispec.docs(tags=['abilities'],
-                          summary='Upload a YAML ability file.',
-                          description='Uploads a YAML ability file, validates its contents, '
-                                      'saves it to disk under data/abilities/{tactic}/, '
-                                      'and loads it into memory.')
-    @aiohttp_apispec.form_schema(AbilityUploadRequestSchema)
-    @aiohttp_apispec.response_schema(AbilitySchema,
-                                     description='JSON dictionary representation of the uploaded Ability.')
-    async def upload_ability(self, request: web.Request):
-        # aiohttp_apispec.form_schema() already parses multipart data and populates request["form"].
-        # Use the pre-parsed form to avoid re-reading the request body which causes multipart boundary errors.
-        file_field: web.FileField = request.get("form", {}).get("file")
-        if not file_field:
-            return json.dumps({'error': 'Missing "file" field in multipart form data.'})
-        filename = file_field.filename or ''
-        # file_field.file is a synchronous file-like object; read it in a thread to avoid blocking the event loop.
-        loop = __import__('asyncio').get_event_loop()
-        file_data = await loop.run_in_executor(None, file_field.file.read)
-        access = await self.get_request_permissions(request)
-        ability = await self._api_manager.upload_ability_file(file_data, filename, access)
-        return web.json_response(ability.display)
diff --git a/app/api/v2/managers/ability_api_manager.py b/app/api/v2/managers/ability_api_manager.py
index f0b3f7dfa..4b0a5504c 100644
--- a/app/api/v2/managers/ability_api_manager.py
+++ b/app/api/v2/managers/ability_api_manager.py
@@ -7,7 +7,7 @@
 
 from app.api.v2.managers.base_api_manager import BaseApiManager
 from app.api.v2.responses import JsonHttpBadRequest
-from app.objects.c_ability import AbilitySchema, Ability
+from app.objects.c_ability import AbilitySchema
 from app.utility.base_world import BaseWorld
 
 
@@ -17,7 +17,9 @@ def __init__(self, data_svc, file_svc):
 
     async def create_on_disk_object(self, data: dict, access: dict, ram_key: str, id_property: str, obj_class: type):
         self._validate_ability_data(create=True, data=data)
-        obj_id = data.get('id')
+        obj_id = data['id']
+        if self.find_object(ram_key, {id_property: obj_id}):
+            raise JsonHttpBadRequest(f'Ability with given id already exists: {obj_id}')
         file_path = self._create_ability_filepath(data.get('tactic'), obj_id)
         allowed = self._get_allowed_from_access(access)
         await self._save_and_reload_object(file_path, data, obj_class, allowed)
@@ -56,6 +58,8 @@ def _validate_ability_data(self, create: bool, data: dict):
             data['id'] = data.pop('ability_id')
         else:
             data['id'] = data.get('id', '')
+        if data['id'] is not None:
+            data['id'] = str(data['id'])
 
         # If a new ability is being created, ensure required fields present.
         if create:
@@ -64,9 +68,9 @@ def _validate_ability_data(self, create: bool, data: dict):
                 data['id'] = str(uuid.uuid4())
             if not data.get('name'):
                 raise JsonHttpBadRequest(f'Cannot create ability {data["id"]} due to missing name')
-            if 'tactic' not in data:
+            if not data.get('tactic'):
                 raise JsonHttpBadRequest(f'Cannot create ability {data["id"]} due to missing tactic')
-            if not data.get('executors'):
+            if not (data.get('executors') or data.get('platforms')):
                 raise JsonHttpBadRequest(f'Cannot create ability {data["id"]}: at least one executor required')
         # Validate ID, used for file creation
         validator = re.compile(r'^[a-zA-Z0-9-_]+$')
@@ -81,7 +85,7 @@ def _validate_ability_data(self, create: bool, data: dict):
                                          'alphanumeric characters, hyphens, and underscores.')
             data['tactic'] = data['tactic'].lower()
 
-        if 'executors' in data and not data.get('executors'):
+        if 'executors' in data and not data.get('executors') and 'platforms' not in data:
             raise JsonHttpBadRequest(f'Cannot create ability {data["id"]}: at least one executor required')
 
         if 'name' in data and not data.get('name'):
@@ -93,40 +97,6 @@ def _create_ability_filepath(self, tactic: str, obj_id: str):
             os.makedirs(tactic_dir)
         return os.path.join(tactic_dir, '%s.yml' % obj_id)
 
-    async def upload_ability_file(self, file_data: bytes, filename: str, access: dict):
-        """Upload a YAML ability file, validate it, save to disk, and load into memory."""
-        # Validate file extension
-        if not filename.lower().endswith(('.yml', '.yaml')):
-            raise JsonHttpBadRequest('Invalid file type. Only .yml and .yaml files are accepted.')
-
-        # Parse YAML
-        try:
-            # Ensure we pass a text string to yaml.safe_load. PyYAML accepts strings or file-like objects.
-            if isinstance(file_data, (bytes, bytearray)):
-                try:
-                    file_text = file_data.decode('utf-8')
-                except UnicodeDecodeError:
-                    # Fallback to latin-1 to preserve byte values if utf-8 fails
-                    file_text = file_data.decode('latin-1')
-            else:
-                file_text = file_data
-            parsed = yaml.safe_load(file_text)
-        except yaml.YAMLError as e:
-            raise JsonHttpBadRequest(f'Invalid YAML: {e}')
-
-        # Handle list-wrapped abilities (common format: list of one dict)
-        if isinstance(parsed, list):
-            if len(parsed) == 0:
-                raise JsonHttpBadRequest('YAML file contains an empty list.')
-            parsed = parsed[0]
-
-        if not isinstance(parsed, dict):
-            raise JsonHttpBadRequest('YAML file must contain a mapping/dictionary.')
-
-        # write the file and return the created ability object
-        return await self.create_on_disk_object(data=parsed, access=access,
-                                                ram_key='abilities', id_property='ability_id', obj_class=Ability)
-
     async def _save_and_reload_object(self, file_path: str, data: dict, obj_type: type, access: BaseWorld.Access):
         await self._file_svc.save_file(file_path, yaml.dump([data], encoding='utf-8', sort_keys=False),
                                        '', encrypt=False)
diff --git a/app/api/v2/schemas/ability_schemas.py b/app/api/v2/schemas/ability_schemas.py
deleted file mode 100644
index 0a4f53ce9..000000000
--- a/app/api/v2/schemas/ability_schemas.py
+++ /dev/null
@@ -1,38 +0,0 @@
-from marshmallow import fields, schema
-
-
-class AbilityUploadRequestSchema(schema.Schema):
-    """Schema for the multipart upload endpoint expecting a YAML file containing an ability.
-
-    The `file` field is a raw file upload. The metadata contains a short description and an
-    example YAML payload that matches the tests in `tests/api/v2/handlers/test_ability_upload.py`.
-    """
-
-    file = fields.Raw(required=True, metadata={
-        'type': 'file',
-        'description': 'A YAML file containing one ability (or a list with one ability). The tests send a YAML list with a single ability entry.',
-        'example': (
-            "- id: upload-test-001\n"
-            "  name: Uploaded Test Ability\n"
-            "  description: An ability uploaded via YAML file\n"
-            "  tactic: discovery\n"
-            "  technique_id: T1082\n"
-            "  technique_name: System Information Discovery\n"
-            "  executors:\n"
-            "  - name: sh\n"
-            "    platform: linux\n"
-            "    command: uname -a\n"
-        ),
-        'example_alternative': (
-            "- ability_id: upload-test-002\n"
-            "  name: Uploaded Test Ability 2\n"
-            "  description: An ability using ability_id key\n"
-            "  tactic: collection\n"
-            "  technique_id: T1005\n"
-            "  technique_name: Data from Local System\n"
-            "  executors:\n"
-            "  - name: sh\n"
-            "    platform: linux\n"
-            "    command: ls -la\n"
-        )
-    })
diff --git a/app/objects/c_ability.py b/app/objects/c_ability.py
index 188c133c0..993010577 100644
--- a/app/objects/c_ability.py
+++ b/app/objects/c_ability.py
@@ -34,17 +34,67 @@ class Meta:
     delete_payload = ma.fields.Bool(load_default=None)
 
     @ma.pre_load
-    def fix_id(self, data, **_):
+    def normalize_ability_file_fields(self, data, **_):
+        if not isinstance(data, dict):
+            return data
         if 'id' in data:
             data['ability_id'] = data.pop('id')
+        if isinstance(data.get('technique'), dict):
+            technique = data.pop('technique')
+            data.setdefault('technique_id', technique.get('attack_id'))
+            data.setdefault('technique_name', technique.get('name'))
+        if 'platforms' in data and 'executors' not in data:
+            data['executors'] = self._platforms_to_executor_list(data.pop('platforms'))
+        if self._has_legacy_requirements(data.get('requirements')):
+            data['requirements'] = self._legacy_requirements_to_list(data['requirements'])
         return data
 
     @ma.post_load
     def build_ability(self, data, **kwargs):
-        if 'technique' in data:
-            data['technique_name'] = data.pop('technique')
         return None if kwargs.get('partial') is True else Ability(**data)
 
+    @staticmethod
+    def _platforms_to_executor_list(platforms):
+        executors = []
+        if not isinstance(platforms, dict):
+            return executors
+        for platform_names, platform_executors in platforms.items():
+            if not isinstance(platform_executors, dict):
+                continue
+            for executor_names, executor_data in platform_executors.items():
+                executor = dict(executor_data or {})
+                if isinstance(executor.get('cleanup'), str):
+                    executor['cleanup'] = [executor['cleanup']]
+                if isinstance(executor.get('parsers'), dict):
+                    executor['parsers'] = [
+                        dict(module=module, parserconfigs=parserconfigs)
+                        for module, parserconfigs in executor['parsers'].items()
+                    ]
+                for platform_name in str(platform_names).split(','):
+                    for executor_name in str(executor_names).split(','):
+                        parsed_executor = dict(executor)
+                        parsed_executor['platform'] = platform_name.strip()
+                        parsed_executor['name'] = executor_name.strip()
+                        executors.append(parsed_executor)
+        return executors
+
+    @staticmethod
+    def _has_legacy_requirements(requirements):
+        return (
+            isinstance(requirements, list)
+            and requirements
+            and isinstance(requirements[0], dict)
+            and 'relationship_match' not in requirements[0]
+        )
+
+    @staticmethod
+    def _legacy_requirements_to_list(requirements):
+        converted = []
+        for requirement in requirements:
+            for module, relationship_match in requirement.items():
+                converted.append(dict(module=module, relationship_match=relationship_match))
+        return converted
+
 
 class Ability(FirstClassObjectInterface, BaseObject):
 
diff --git a/tests/api/v2/handlers/test_ability_upload.py b/tests/api/v2/handlers/test_ability_upload.py
index 113b1974f..04b77fe40 100644
--- a/tests/api/v2/handlers/test_ability_upload.py
+++ b/tests/api/v2/handlers/test_ability_upload.py
@@ -1,92 +1,111 @@
 import os
 import pytest
-import yaml
 
-from aiohttp import FormData
 from http import HTTPStatus
 
+from app.utility.base_service import BaseService
 
-@pytest.fixture
-def valid_ability_yaml():
-    ability_data = {
-        'id': 'upload-test-001',
-        'name': 'Uploaded Test Ability',
-        'description': 'An ability uploaded via YAML file',
-        'tactic': 'discovery',
-        'technique_id': 'T1082',
-        'technique_name': 'System Information Discovery',
-        'executors': [
-            {
-                'name': 'sh',
-                'platform': 'linux',
-                'command': 'uname -a'
-            }
-        ]
-    }
-    yield yaml.dump([ability_data], sort_keys=False).encode('utf-8')
 
-    # Cleanup
-    path = 'data/abilities/discovery/upload-test-001.yml'
-    if os.path.exists(path):
+def ability_file_cleanup(tactic, ability_id):
+    file_path = f'data/abilities/{tactic}/{ability_id}.yml'
+    if os.path.exists(file_path):
         try:
-            os.remove(path)
+            os.remove(file_path)
         except OSError:
             pass
 
 
-@pytest.fixture
-def valid_ability_yaml_with_ability_id_key():
-    """Uses 'ability_id' key instead of 'id'."""
-    ability_data = {
-        'ability_id': 'upload-test-002',
-        'name': 'Uploaded Test Ability 2',
-        'description': 'An ability using ability_id key',
-        'tactic': 'collection',
-        'technique_id': 'T1005',
-        'technique_name': 'Data from Local System',
-        'executors': [
+def basic_ability(identifier_key, identifier, name, description, tactic):
+    return {
+        identifier_key: identifier,
+        'name': name,
+        'description': description,
+        'tactic': tactic,
+        'technique': {
+            'attack_id': 'T1083',
+            'name': 'File and Directory Discovery'
+        },
+        'platforms': {
+            'darwin': {
+                'sh': {
+                    'command': 'ls #{host.system.path}'
+                }
+            },
+            'linux': {
+                'sh': {
+                    'command': 'ls #{host.system.path}'
+                }
+            },
+            'windows': {
+                'psh': {
+                    'command': 'dir #{host.system.path}'
+                }
+            }
+        },
+        'requirements': [
             {
-                'name': 'sh',
-                'platform': 'linux',
-                'command': 'ls -la'
+                'plugins.stockpile.app.requirements.paw_provenance': [
+                    {
+                        'source': 'host.system.path'
+                    }
+                ]
             }
         ]
     }
-    yield yaml.dump([ability_data], sort_keys=False).encode('utf-8')
 
-    path = 'data/abilities/collection/upload-test-002.yml'
-    if os.path.exists(path):
-        try:
-            os.remove(path)
-        except OSError:
-            pass
+
+@pytest.fixture
+def valid_ability_payload():
+    yield basic_ability(
+        'id',
+        'upload-test-001',
+        'Uploaded Test Ability',
+        'An ability uploaded via YAML file',
+        'discovery'
+    )
+
+    ability_file_cleanup('discovery', 'upload-test-001')
+
+
+@pytest.fixture
+def valid_ability_payload_with_ability_id():
+    yield basic_ability(
+        'ability_id',
+        'upload-test-002',
+        'Uploaded Test Ability 2',
+        'An ability using ability_id key',
+        'collection'
+    )
+
+    ability_file_cleanup('collection', 'upload-test-002')
 
 
 class TestAbilityUploadApi:
 
-    async def test_upload_valid_yaml(self, api_v2_client, api_cookies, valid_ability_yaml):
-        form = FormData()
-        form.add_field('file', valid_ability_yaml,
-                       filename='upload-test-001.yml',
-                       content_type='application/x-yaml')
-        resp = await api_v2_client.post('/api/v2/abilities/upload',
-                                        cookies=api_cookies,
-                                        data=form)
+    async def test_create_ability_from_yaml_style_payload(self, api_v2_client, api_cookies, valid_ability_payload):
+        resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=valid_ability_payload)
+
         assert resp.status == HTTPStatus.OK
         result = await resp.json()
         assert result['ability_id'] == 'upload-test-001'
         assert result['name'] == 'Uploaded Test Ability'
         assert result['tactic'] == 'discovery'
+        assert result['technique_id'] == 'T1083'
+        assert result['technique_name'] == 'File and Directory Discovery'
+        assert {executor['platform'] for executor in result['executors']} == {'darwin', 'linux', 'windows'}
         assert os.path.exists('data/abilities/discovery/upload-test-001.yml')
 
-    async def test_upload_valid_yaml_ability_id(self, api_v2_client, api_cookies, valid_ability_yaml_with_ability_id_key):
-        form = FormData()
-        form.add_field('file', valid_ability_yaml_with_ability_id_key,
-                       filename='upload-test-001.yml',
-                       content_type='application/x-yaml')
-        resp = await api_v2_client.post('/api/v2/abilities/upload',
-                                        cookies=api_cookies,
-                                        data=form)
+        ability = (await BaseService.get_service('data_svc').locate(
+            'abilities', {'ability_id': 'upload-test-001'}
+        ))[0]
+        assert ability.requirements[0].module == 'plugins.stockpile.app.requirements.paw_provenance'
+
+    async def test_create_ability_from_yaml_style_payload_with_ability_id(
+            self, api_v2_client, api_cookies, valid_ability_payload_with_ability_id
+    ):
+        resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies,
+                                        json=valid_ability_payload_with_ability_id)
+
         assert resp.status == HTTPStatus.OK
         result = await resp.json()
         assert result['ability_id'] == 'upload-test-002'
@@ -94,42 +113,13 @@ async def test_upload_valid_yaml_ability_id(self, api_v2_client, api_cookies, va
         assert result['tactic'] == 'collection'
         assert os.path.exists('data/abilities/collection/upload-test-002.yml')
 
-    async def test_upload_invalid_file_type(self, api_v2_client, api_cookies):
-        form = FormData()
-        form.add_field('file', b'not yaml content',
-                       filename='bad_file.txt',
-                       content_type='text/plain')
-        resp = await api_v2_client.post('/api/v2/abilities/upload',
-                                        cookies=api_cookies,
-                                        data=form)
-        assert resp.status == HTTPStatus.BAD_REQUEST
+    async def test_create_ability_from_yaml_style_payload_missing_required_fields(self, api_v2_client, api_cookies):
+        resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies,
+                                        json={'description': 'no name, tactic, or executor configuration'})
 
-    async def test_upload_malformed_yaml(self, api_v2_client, api_cookies):
-        form = FormData()
-        form.add_field('file', b'{{invalid: yaml: [}',
-                       filename='malformed.yml',
-                       content_type='application/x-yaml')
-        resp = await api_v2_client.post('/api/v2/abilities/upload',
-                                        cookies=api_cookies,
-                                        data=form)
         assert resp.status == HTTPStatus.BAD_REQUEST
 
-    async def test_upload_missing_required_fields(self, api_v2_client, api_cookies):
-        incomplete = yaml.dump([{'description': 'no id, name, or tactic'}]).encode('utf-8')
-        form = FormData()
-        form.add_field('file', incomplete,
-                       filename='incomplete.yml',
-                       content_type='application/x-yaml')
-        resp = await api_v2_client.post('/api/v2/abilities/upload',
-                                        cookies=api_cookies,
-                                        data=form)
-        assert resp.status == HTTPStatus.BAD_REQUEST
+    async def test_unauthorized_create_ability_from_yaml_style_payload(self, api_v2_client, valid_ability_payload):
+        resp = await api_v2_client.post('/api/v2/abilities', json=valid_ability_payload)
 
-    async def test_unauthorized_upload(self, api_v2_client, valid_ability_yaml):
-        form = FormData()
-        form.add_field('file', valid_ability_yaml,
-                       filename='upload-test-001.yml',
-                       content_type='application/x-yaml')
-        resp = await api_v2_client.post('/api/v2/abilities/upload',
-                                        data=form)
         assert resp.status == HTTPStatus.UNAUTHORIZED

From 087cbd901fe07340830c125ff06ecab8ca443c9a Mon Sep 17 00:00:00 2001
From: Fiona McCrae <fmccrae@mitre.org>
Date: Thu, 30 Apr 2026 17:31:15 -0400
Subject: [PATCH 07/10] attempted to simplify platform/executor code and added
 tests to reflect new YAML formats

---
 app/objects/c_ability.py                     | 26 ++++--
 tests/api/v2/handlers/test_ability_upload.py | 98 +++++++++++++++++++-
 2 files changed, 111 insertions(+), 13 deletions(-)

diff --git a/app/objects/c_ability.py b/app/objects/c_ability.py
index 993010577..7a96e330c 100644
--- a/app/objects/c_ability.py
+++ b/app/objects/c_ability.py
@@ -35,6 +35,9 @@ class Meta:
 
     @ma.pre_load
     def normalize_ability_file_fields(self, data, **_):
+        """
+        Ensures that ability file fields are formatted correctly for processing
+        """
         if not isinstance(data, dict):
             return data
         if 'id' in data:
@@ -55,27 +58,34 @@ def build_ability(self, data, **kwargs):
 
     @staticmethod
     def _platforms_to_executor_list(platforms):
+        """
+        Translates legacy platform-structured YAML into caldera executor format
+        """
         executors = []
         if not isinstance(platforms, dict):
+            # there are no executors listed, return empty list
             return executors
         for platform_names, platform_executors in platforms.items():
             if not isinstance(platform_executors, dict):
+                # there is just one executor for this platform
                 continue
+            platform_list = [name.strip() for name in str(platform_names).split(',')]
             for executor_names, executor_data in platform_executors.items():
-                executor = dict(executor_data or {})
+                executor = dict(executor_data)  # make a dict of the data and fix up below
                 if isinstance(executor.get('cleanup'), str):
+                    # cleanup actions should be in a list
                     executor['cleanup'] = [executor['cleanup']]
                 if isinstance(executor.get('parsers'), dict):
                     executor['parsers'] = [
-                        dict(module=module, parserconfigs=parserconfigs)
+                        {'module': module, 'parserconfigs': parserconfigs}
                         for module, parserconfigs in executor['parsers'].items()
                     ]
-                for platform_name in str(platform_names).split(','):
-                    for executor_name in str(executor_names).split(','):
-                        parsed_executor = dict(executor)
-                        parsed_executor['platform'] = platform_name.strip()
-                        parsed_executor['name'] = executor_name.strip()
-                        executors.append(parsed_executor)
+                executor_list = [name.strip() for name in str(executor_names.split(','))]
+                executors.extend(
+                    {**executor, 'platform': platform_name, 'name': executor_name}
+                    for platform_name in platform_list
+                    for executor_name in executor_list
+                )
         return executors
 
     @staticmethod
diff --git a/tests/api/v2/handlers/test_ability_upload.py b/tests/api/v2/handlers/test_ability_upload.py
index 04b77fe40..adde8f252 100644
--- a/tests/api/v2/handlers/test_ability_upload.py
+++ b/tests/api/v2/handlers/test_ability_upload.py
@@ -15,7 +15,7 @@ def ability_file_cleanup(tactic, ability_id):
             pass
 
 
-def basic_ability(identifier_key, identifier, name, description, tactic):
+def basic_platform_ability(identifier_key, identifier, name, description, tactic):
     return {
         identifier_key: identifier,
         'name': name,
@@ -56,7 +56,7 @@ def basic_ability(identifier_key, identifier, name, description, tactic):
 
 @pytest.fixture
 def valid_ability_payload():
-    yield basic_ability(
+    yield basic_platform_ability(
         'id',
         'upload-test-001',
         'Uploaded Test Ability',
@@ -69,7 +69,7 @@ def valid_ability_payload():
 
 @pytest.fixture
 def valid_ability_payload_with_ability_id():
-    yield basic_ability(
+    yield basic_platform_ability(
         'ability_id',
         'upload-test-002',
         'Uploaded Test Ability 2',
@@ -80,9 +80,69 @@ def valid_ability_payload_with_ability_id():
     ability_file_cleanup('collection', 'upload-test-002')
 
 
+@pytest.fixture
+def new_executors_ability_payload():
+    ability_id = 'upload-test-new-executors'
+    tactic = 'discovery'
+    ability_file_cleanup(tactic, ability_id)
+    yield {
+        'id': ability_id,
+        'repeatable': False,
+        'name': 'New executors ability',
+        'additional_info': {
+            'cleanup': ''
+        },
+        'technique_name': 'File and Directory Discovery',
+        'executors': [
+            {
+                'name': 'sh',
+                'additional_info': {},
+                'variations': [],
+                'platform': 'linux',
+                'command': 'ls',
+                'code': None,
+                'language': None,
+                'payloads': [],
+                'timeout': 60,
+                'parsers': [],
+                'cleanup': [],
+                'uploads': [],
+                'build_target': None
+            },
+            {
+                'name': 'psh',
+                'additional_info': {},
+                'variations': [],
+                'platform': 'windows',
+                'command': 'dir',
+                'code': None,
+                'language': None,
+                'payloads': [],
+                'timeout': 60,
+                'parsers': [],
+                'cleanup': [],
+                'uploads': [],
+                'build_target': None
+            }
+        ],
+        'buckets': [],
+        'technique_id': 'T1083',
+        'delete_payload': True,
+        'tactic': tactic,
+        'description': 'Simple new-style ability payload.',
+        'singleton': False,
+        'plugin': '',
+        'requirements': [],
+        'privilege': '',
+        'access': {}
+    }
+    ability_file_cleanup(tactic, ability_id)
+
+
 class TestAbilityUploadApi:
 
-    async def test_create_ability_from_yaml_style_payload(self, api_v2_client, api_cookies, valid_ability_payload):
+    async def test_create_ability_from_old_platforms_yaml_style_payload(self, api_v2_client, api_cookies,
+                                                                        valid_ability_payload):
         resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=valid_ability_payload)
 
         assert resp.status == HTTPStatus.OK
@@ -92,12 +152,20 @@ async def test_create_ability_from_yaml_style_payload(self, api_v2_client, api_c
         assert result['tactic'] == 'discovery'
         assert result['technique_id'] == 'T1083'
         assert result['technique_name'] == 'File and Directory Discovery'
-        assert {executor['platform'] for executor in result['executors']} == {'darwin', 'linux', 'windows'}
+        assert [
+            (executor['platform'], executor['name'], executor['command'])
+            for executor in result['executors']
+        ] == [
+            ('darwin', 'sh', 'ls #{host.system.path}'),
+            ('linux', 'sh', 'ls #{host.system.path}'),
+            ('windows', 'psh', 'dir #{host.system.path}')
+        ]
         assert os.path.exists('data/abilities/discovery/upload-test-001.yml')
 
         ability = (await BaseService.get_service('data_svc').locate(
             'abilities', {'ability_id': 'upload-test-001'}
         ))[0]
+        assert ability.display == result
         assert ability.requirements[0].module == 'plugins.stockpile.app.requirements.paw_provenance'
 
     async def test_create_ability_from_yaml_style_payload_with_ability_id(
@@ -113,6 +181,26 @@ async def test_create_ability_from_yaml_style_payload_with_ability_id(
         assert result['tactic'] == 'collection'
         assert os.path.exists('data/abilities/collection/upload-test-002.yml')
 
+    async def test_create_ability_from_new_executors_yaml_style(self, api_v2_client, api_cookies,
+                                                                new_executors_ability_payload):
+        resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies,
+                                        json=new_executors_ability_payload)
+
+        assert resp.status == HTTPStatus.OK
+        ability_data = await resp.json()
+        assert ability_data['ability_id'] == new_executors_ability_payload['id']
+        assert ability_data['technique_id'] == new_executors_ability_payload['technique_id']
+        assert ability_data['technique_name'] == new_executors_ability_payload['technique_name']
+        assert [
+            (executor['platform'], executor['name'], executor['command'])
+            for executor in ability_data['executors']
+        ] == [('linux', 'sh', 'ls'), ('windows', 'psh', 'dir')]
+
+        stored_ability = (await BaseService.get_service('data_svc').locate(
+            'abilities', {'ability_id': new_executors_ability_payload['id']}
+        ))[0]
+        assert stored_ability.display == ability_data
+
     async def test_create_ability_from_yaml_style_payload_missing_required_fields(self, api_v2_client, api_cookies):
         resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies,
                                         json={'description': 'no name, tactic, or executor configuration'})

From 5a0a8fc17d74835258fe4742d2e884e62299071e Mon Sep 17 00:00:00 2001
From: Fiona McCrae <fmccrae@mitre.org>
Date: Wed, 6 May 2026 15:15:31 -0400
Subject: [PATCH 08/10] added additional validation steps

---
 app/api/v2/managers/ability_api_manager.py   | 107 ++++++++++-
 app/objects/c_ability.py                     |  15 +-
 tests/api/v2/handlers/test_ability_upload.py | 184 +++++++++++++++++++
 3 files changed, 293 insertions(+), 13 deletions(-)

diff --git a/app/api/v2/managers/ability_api_manager.py b/app/api/v2/managers/ability_api_manager.py
index 4b0a5504c..e36dd015d 100644
--- a/app/api/v2/managers/ability_api_manager.py
+++ b/app/api/v2/managers/ability_api_manager.py
@@ -8,10 +8,13 @@
 from app.api.v2.managers.base_api_manager import BaseApiManager
 from app.api.v2.responses import JsonHttpBadRequest
 from app.objects.c_ability import AbilitySchema
+from app.service.file_svc import FileSvc
 from app.utility.base_world import BaseWorld
 
 
 class AbilityApiManager(BaseApiManager):
+    _EXECUTOR_LABEL_PATTERN = re.compile(r'^[a-zA-Z0-9_.-]+$')
+
     def __init__(self, data_svc, file_svc):
         super().__init__(data_svc=data_svc, file_svc=file_svc)
 
@@ -60,7 +63,6 @@ def _validate_ability_data(self, create: bool, data: dict):
             data['id'] = data.get('id', '')
         if data['id'] is not None:
             data['id'] = str(data['id'])
-
         # If a new ability is being created, ensure required fields present.
         if create:
             # Set ability ID if undefined
@@ -72,15 +74,13 @@ def _validate_ability_data(self, create: bool, data: dict):
                 raise JsonHttpBadRequest(f'Cannot create ability {data["id"]} due to missing tactic')
             if not (data.get('executors') or data.get('platforms')):
                 raise JsonHttpBadRequest(f'Cannot create ability {data["id"]}: at least one executor required')
-        # Validate ID, used for file creation
-        validator = re.compile(r'^[a-zA-Z0-9-_]+$')
-        if 'id' in data and not validator.match(data['id']):
-            raise JsonHttpBadRequest(f'Invalid ability ID {data["id"]}. IDs can only contain '
-                                     'alphanumeric characters, hyphens, and underscores.')
+        # Sanitize ID, used for file creation
+        data['id'] = BaseApiManager._sanitize_id(data['id'])
 
         # Validate tactic, used for directory creation, lower case if present
+        validator = re.compile(r'^[a-zA-Z0-9-_]+$')
         if 'tactic' in data:
-            if not validator.match(data['tactic']):
+            if not isinstance(data['tactic'], str) or not validator.match(data['tactic']):
                 raise JsonHttpBadRequest(f'Invalid ability tactic {data["tactic"]}. Tactics can only contain '
                                          'alphanumeric characters, hyphens, and underscores.')
             data['tactic'] = data['tactic'].lower()
@@ -91,6 +91,99 @@ def _validate_ability_data(self, create: bool, data: dict):
         if 'name' in data and not data.get('name'):
             raise JsonHttpBadRequest(f'Cannot create ability {data["id"]} due to missing name')
 
+        self._validate_ability_privilege(data)
+        self._validate_ability_executors(data)
+
+    def _validate_ability_privilege(self, data: dict):
+        if 'privilege' not in data:
+            return
+
+        privilege = data.get('privilege')
+        if privilege is None or privilege == '':
+            return
+        if not isinstance(privilege, str):
+            raise JsonHttpBadRequest(f'Invalid ability privilege {privilege}. Privilege must be one of: '
+                                     'User, Elevated.')
+
+        allowed_privileges = {privilege.name for privilege in BaseWorld.Privileges}
+        if privilege not in allowed_privileges:
+            raise JsonHttpBadRequest(f'Invalid ability privilege {privilege}. Privilege must be one of: '
+                                     'User, Elevated.')
+
+    def _validate_ability_executors(self, data: dict):
+        if data.get('executors') is not None:
+            self._validate_executor_list(data['executors'])
+        if data.get('platforms') is not None:
+            self._validate_platform_executor_map(data['platforms'])
+
+    def _validate_executor_list(self, executors):
+        if not isinstance(executors, list):
+            raise JsonHttpBadRequest('Invalid ability executors. Executors must be a list.')
+
+        for index, executor in enumerate(executors):
+            if not isinstance(executor, dict):
+                raise JsonHttpBadRequest(f'Invalid ability executor at index {index}. Executor must be a dictionary.')
+            self._validate_executor_label(executor.get('name'), f'executor[{index}].name')
+            self._validate_executor_label(executor.get('platform'), f'executor[{index}].platform')
+            self._validate_payloads(executor.get('payloads'), f'executor[{index}].payloads')
+
+    def _validate_platform_executor_map(self, platforms):
+        if not isinstance(platforms, dict):
+            raise JsonHttpBadRequest('Invalid ability platforms. Platforms must be a dictionary.')
+
+        for platform_names, platform_executors in platforms.items():
+            for platform_name in self._split_and_validate_labels(platform_names, 'platform'):
+                if not isinstance(platform_executors, dict):
+                    raise JsonHttpBadRequest(f'Invalid ability platform {platform_name}. Platform executors must be '
+                                             'a dictionary.')
+                for executor_names, executor in platform_executors.items():
+                    for executor_name in self._split_and_validate_labels(executor_names, 'executor'):
+                        if not isinstance(executor, dict):
+                            raise JsonHttpBadRequest(f'Invalid ability executor {executor_name} for platform '
+                                                     f'{platform_name}. Executor must be a dictionary.')
+                    self._validate_payloads(executor.get('payloads'), f'platforms.{platform_name}.payloads')
+
+    @classmethod
+    def _split_and_validate_labels(cls, value, field_name):
+        if not isinstance(value, str):
+            raise JsonHttpBadRequest(f'Invalid ability {field_name} {value}. {field_name.capitalize()} names must be '
+                                     'strings.')
+
+        labels = [label.strip() for label in value.split(',')]
+        if not labels or any(not label for label in labels):
+            raise JsonHttpBadRequest(f'Invalid ability {field_name} {value}. {field_name.capitalize()} names cannot '
+                                     'be empty.')
+
+        for label in labels:
+            cls._validate_executor_label(label, field_name)
+        return labels
+
+    @classmethod
+    def _validate_executor_label(cls, value, field_name):
+        if not isinstance(value, str) or not value:
+            raise JsonHttpBadRequest(f'Invalid ability {field_name}. Executor and platform names must be non-empty '
+                                     'strings.')
+        if not cls._EXECUTOR_LABEL_PATTERN.match(value):
+            raise JsonHttpBadRequest(f'Invalid ability {field_name} {value}. Executor and platform names can only '
+                                     'contain alphanumeric characters, periods, hyphens, and underscores.')
+
+    @staticmethod
+    def _validate_payloads(payloads, field_name):
+        if payloads is None:
+            return
+        if not isinstance(payloads, list):
+            raise JsonHttpBadRequest(f'Invalid ability {field_name}. Payloads must be a list.')
+
+        for payload in payloads:
+            if not isinstance(payload, str):
+                raise JsonHttpBadRequest(f'Invalid ability payload {payload}. Payload names must be strings.')
+            safe_filename = FileSvc._validate_filename(payload)
+            if BaseWorld.is_uuid4(payload) and safe_filename:
+                continue
+            if not safe_filename:
+                raise JsonHttpBadRequest(f'Invalid ability payload {payload}. Payload names cannot contain path '
+                                         'separators, traversal sequences, null bytes, or unsafe characters.')
+
     def _create_ability_filepath(self, tactic: str, obj_id: str):
         tactic_dir = os.path.join('data', 'abilities', tactic)
         if not os.path.exists(tactic_dir):
diff --git a/app/objects/c_ability.py b/app/objects/c_ability.py
index 7a96e330c..00e325b54 100644
--- a/app/objects/c_ability.py
+++ b/app/objects/c_ability.py
@@ -63,14 +63,16 @@ def _platforms_to_executor_list(platforms):
         """
         executors = []
         if not isinstance(platforms, dict):
-            # there are no executors listed, return empty list
-            return executors
+            raise ma.ValidationError('Platforms must be a dictionary.', 'platforms')
         for platform_names, platform_executors in platforms.items():
             if not isinstance(platform_executors, dict):
-                # there is just one executor for this platform
-                continue
+                raise ma.ValidationError('Platform executors must be a dictionary.', 'platforms')
+
             platform_list = [name.strip() for name in str(platform_names).split(',')]
+
             for executor_names, executor_data in platform_executors.items():
+                if not isinstance(executor_data, dict):
+                    raise ma.ValidationError('Executor data must be a dictionary.', 'platforms')
                 executor = dict(executor_data)  # make a dict of the data and fix up below
                 if isinstance(executor.get('cleanup'), str):
                     # cleanup actions should be in a list
@@ -80,7 +82,8 @@ def _platforms_to_executor_list(platforms):
                         {'module': module, 'parserconfigs': parserconfigs}
                         for module, parserconfigs in executor['parsers'].items()
                     ]
-                executor_list = [name.strip() for name in str(executor_names.split(','))]
+
+                executor_list = [name.strip() for name in str(executor_names).split(',')]
                 executors.extend(
                     {**executor, 'platform': platform_name, 'name': executor_name}
                     for platform_name in platform_list
@@ -102,7 +105,7 @@ def _legacy_requirements_to_list(requirements):
         converted = []
         for requirement in requirements:
             for module, relationship_match in requirement.items():
-                converted.append(dict(module=module, relationship_match=relationship_match))
+                converted.append({'module': module, 'relationship_match': relationship_match})
         return converted
 
 
diff --git a/tests/api/v2/handlers/test_ability_upload.py b/tests/api/v2/handlers/test_ability_upload.py
index adde8f252..cd2926ac1 100644
--- a/tests/api/v2/handlers/test_ability_upload.py
+++ b/tests/api/v2/handlers/test_ability_upload.py
@@ -1,3 +1,4 @@
+import copy
 import os
 import pytest
 
@@ -54,6 +55,13 @@ def basic_platform_ability(identifier_key, identifier, name, description, tactic
     }
 
 
+def cleanup_payload_ability(payload):
+    ability_id = payload.get('id') or payload.get('ability_id')
+    tactic = payload.get('tactic')
+    if ability_id and tactic:
+        ability_file_cleanup(tactic, ability_id)
+
+
 @pytest.fixture
 def valid_ability_payload():
     yield basic_platform_ability(
@@ -201,6 +209,182 @@ async def test_create_ability_from_new_executors_yaml_style(self, api_v2_client,
         ))[0]
         assert stored_ability.display == ability_data
 
+    async def test_create_ability_accepts_safe_payload_and_plugin_style_executor_names(
+            self, api_v2_client, api_cookies, new_executors_ability_payload
+    ):
+        payload = copy.deepcopy(new_executors_ability_payload)
+        payload['id'] = 'upload-test-plugin-executor-labels'
+        payload['executors'][0]['name'] = 'plugin.exec-1'
+        payload['executors'][0]['payloads'] = [
+            'safe-payload.ps1',
+            '766be199-7316-4b26-b3db-e272aaf7e0d4'
+        ]
+
+        try:
+            resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=payload)
+
+            assert resp.status == HTTPStatus.OK
+            ability_data = await resp.json()
+            assert ability_data['executors'][0]['name'] == 'plugin.exec-1'
+            assert ability_data['executors'][0]['payloads'] == payload['executors'][0]['payloads']
+        finally:
+            cleanup_payload_ability(payload)
+
+    @pytest.mark.parametrize(
+        ('payload_name', 'suffix'),
+        [
+            ('../evil.ps1', 'parent'),
+            ('payloads/evil.ps1', 'nested'),
+            ('/tmp/evil.ps1', 'absolute'),
+            ('evil\x00.ps1', 'null-byte'),
+            ('evil;rm.ps1', 'unsafe'),
+        ]
+    )
+    async def test_create_ability_rejects_unsafe_payload_paths(
+            self, api_v2_client, api_cookies, new_executors_ability_payload, payload_name, suffix
+    ):
+        payload = copy.deepcopy(new_executors_ability_payload)
+        payload['id'] = f'upload-test-invalid-payload-{suffix}'
+        payload['executors'][0]['payloads'] = [payload_name]
+
+        resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=payload)
+
+        assert resp.status == HTTPStatus.BAD_REQUEST
+        assert not os.path.exists(f'data/abilities/{payload["tactic"]}/{payload["id"]}.yml')
+
+    @pytest.mark.parametrize(
+        ('executors', 'suffix'),
+        [
+            (['sh'], 'non-dict'),
+            ([{'platform': 'linux', 'command': 'ls'}], 'missing-name'),
+            ([{'name': 'sh', 'command': 'ls'}], 'missing-platform'),
+            ([{'name': 'sh', 'platform': 'linux', 'command': 'ls', 'payloads': 'payload.ps1'}], 'bad-payloads'),
+        ]
+    )
+    async def test_create_ability_rejects_schema_invalid_new_style_executors(
+            self, api_v2_client, api_cookies, new_executors_ability_payload, executors, suffix
+    ):
+        payload = copy.deepcopy(new_executors_ability_payload)
+        payload['id'] = f'upload-test-schema-invalid-executor-{suffix}'
+        payload['executors'] = executors
+
+        resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=payload)
+
+        assert resp.status == HTTPStatus.UNPROCESSABLE_ENTITY
+        assert not os.path.exists(f'data/abilities/{payload["tactic"]}/{payload["id"]}.yml')
+
+    @pytest.mark.parametrize(
+        ('executors', 'suffix'),
+        [
+            ([{'name': 'sh/evil', 'platform': 'linux', 'command': 'ls'}], 'unsafe-name'),
+            ([{'name': 'sh', 'platform': 'lin ux', 'command': 'ls'}], 'unsafe-platform'),
+        ]
+    )
+    async def test_create_ability_rejects_policy_invalid_new_style_executors(
+            self, api_v2_client, api_cookies, new_executors_ability_payload, executors, suffix
+    ):
+        payload = copy.deepcopy(new_executors_ability_payload)
+        payload['id'] = f'upload-test-policy-invalid-executor-{suffix}'
+        payload['executors'] = executors
+
+        resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=payload)
+
+        assert resp.status == HTTPStatus.BAD_REQUEST
+        assert not os.path.exists(f'data/abilities/{payload["tactic"]}/{payload["id"]}.yml')
+
+    @pytest.mark.parametrize(
+        ('platforms', 'suffix'),
+        [
+            ({'linux': ['sh']}, 'platform-not-dict'),
+            ({'linux': {'sh': 'ls'}}, 'executor-not-dict'),
+        ]
+    )
+    async def test_create_ability_rejects_schema_invalid_legacy_platform_executors(
+            self, api_v2_client, api_cookies, valid_ability_payload, platforms, suffix
+    ):
+        payload = copy.deepcopy(valid_ability_payload)
+        payload['id'] = f'upload-test-schema-invalid-platform-{suffix}'
+        payload['platforms'] = platforms
+
+        resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=payload)
+
+        assert resp.status == HTTPStatus.UNPROCESSABLE_ENTITY
+        assert not os.path.exists(f'data/abilities/{payload["tactic"]}/{payload["id"]}.yml')
+
+    @pytest.mark.parametrize(
+        ('platforms', 'suffix'),
+        [
+            ({'lin/ux': {'sh': {'command': 'ls'}}}, 'unsafe-platform'),
+            ({'linux': {'sh/evil': {'command': 'ls'}}}, 'unsafe-executor'),
+        ]
+    )
+    async def test_create_ability_rejects_policy_invalid_legacy_platform_executors(
+            self, api_v2_client, api_cookies, valid_ability_payload, platforms, suffix
+    ):
+        payload = copy.deepcopy(valid_ability_payload)
+        payload['id'] = f'upload-test-policy-invalid-platform-{suffix}'
+        payload['platforms'] = platforms
+
+        resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=payload)
+
+        assert resp.status == HTTPStatus.BAD_REQUEST
+        assert not os.path.exists(f'data/abilities/{payload["tactic"]}/{payload["id"]}.yml')
+
+    @pytest.mark.parametrize(
+        ('privilege', 'suffix'),
+        [
+            (None, 'none'),
+            ('', 'empty'),
+            ('User', 'user'),
+            ('Elevated', 'elevated'),
+        ]
+    )
+    async def test_create_ability_accepts_valid_privileges(
+            self, api_v2_client, api_cookies, new_executors_ability_payload, privilege, suffix
+    ):
+        payload = copy.deepcopy(new_executors_ability_payload)
+        payload['id'] = f'upload-test-valid-privilege-{suffix}'
+        payload['privilege'] = privilege
+
+        try:
+            resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=payload)
+
+            assert resp.status == HTTPStatus.OK
+        finally:
+            cleanup_payload_ability(payload)
+
+    @pytest.mark.parametrize(
+        ('privilege', 'suffix'),
+        [
+            ('Admin', 'admin'),
+            ('root', 'root'),
+            ('elevated', 'lowercase-elevated'),
+        ]
+    )
+    async def test_create_ability_rejects_policy_invalid_privileges(
+            self, api_v2_client, api_cookies, new_executors_ability_payload, privilege, suffix
+    ):
+        payload = copy.deepcopy(new_executors_ability_payload)
+        payload['id'] = f'upload-test-invalid-privilege-{suffix}'
+        payload['privilege'] = privilege
+
+        resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=payload)
+
+        assert resp.status == HTTPStatus.BAD_REQUEST
+        assert not os.path.exists(f'data/abilities/{payload["tactic"]}/{payload["id"]}.yml')
+
+    async def test_create_ability_rejects_schema_invalid_privilege_type(
+            self, api_v2_client, api_cookies, new_executors_ability_payload
+    ):
+        payload = copy.deepcopy(new_executors_ability_payload)
+        payload['id'] = 'upload-test-schema-invalid-privilege-non-string'
+        payload['privilege'] = 7
+
+        resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=payload)
+
+        assert resp.status == HTTPStatus.UNPROCESSABLE_ENTITY
+        assert not os.path.exists(f'data/abilities/{payload["tactic"]}/{payload["id"]}.yml')
+
     async def test_create_ability_from_yaml_style_payload_missing_required_fields(self, api_v2_client, api_cookies):
         resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies,
                                         json={'description': 'no name, tactic, or executor configuration'})

From ecab1219a529c8801d515089474190743d598146 Mon Sep 17 00:00:00 2001
From: Fiona McCrae <fmccrae@mitre.org>
Date: Wed, 6 May 2026 15:52:50 -0400
Subject: [PATCH 09/10] update to sanitize ID before storing internally

---
 app/api/v2/managers/ability_api_manager.py   | 20 ++++++-----
 tests/api/v2/handlers/test_ability_upload.py | 36 ++++++++++++++++++++
 2 files changed, 47 insertions(+), 9 deletions(-)

diff --git a/app/api/v2/managers/ability_api_manager.py b/app/api/v2/managers/ability_api_manager.py
index e36dd015d..80437ea90 100644
--- a/app/api/v2/managers/ability_api_manager.py
+++ b/app/api/v2/managers/ability_api_manager.py
@@ -57,25 +57,27 @@ async def update_on_disk_object(self, obj: Any, data: dict, ram_key: str, id_pro
 
     def _validate_ability_data(self, create: bool, data: dict):
         # Normalize ability ID: prefer explicit 'ability_id' if provided, otherwise preserve any existing 'id'.
+        ability_id = None
         if 'ability_id' in data:
-            data['id'] = data.pop('ability_id')
+            ability_id = data.pop('ability_id')
+        elif 'id' in data:
+            ability_id = data.get('id')
+
+        # Sanitize supplied IDs before assigning them internally. If no ID is supplied during creation,
+        # generate one instead.
+        if ability_id in (None, '') and create:
+            data['id'] = str(uuid.uuid4())
         else:
-            data['id'] = data.get('id', '')
-        if data['id'] is not None:
-            data['id'] = str(data['id'])
+            data['id'] = BaseApiManager._sanitize_id(ability_id)
+
         # If a new ability is being created, ensure required fields present.
         if create:
-            # Set ability ID if undefined
-            if not data['id']:
-                data['id'] = str(uuid.uuid4())
             if not data.get('name'):
                 raise JsonHttpBadRequest(f'Cannot create ability {data["id"]} due to missing name')
             if not data.get('tactic'):
                 raise JsonHttpBadRequest(f'Cannot create ability {data["id"]} due to missing tactic')
             if not (data.get('executors') or data.get('platforms')):
                 raise JsonHttpBadRequest(f'Cannot create ability {data["id"]}: at least one executor required')
-        # Sanitize ID, used for file creation
-        data['id'] = BaseApiManager._sanitize_id(data['id'])
 
         # Validate tactic, used for directory creation, lower case if present
         validator = re.compile(r'^[a-zA-Z0-9-_]+$')
diff --git a/tests/api/v2/handlers/test_ability_upload.py b/tests/api/v2/handlers/test_ability_upload.py
index cd2926ac1..c1bed7dcf 100644
--- a/tests/api/v2/handlers/test_ability_upload.py
+++ b/tests/api/v2/handlers/test_ability_upload.py
@@ -209,6 +209,42 @@ async def test_create_ability_from_new_executors_yaml_style(self, api_v2_client,
         ))[0]
         assert stored_ability.display == ability_data
 
+    async def test_create_ability_sanitizes_upload_id_before_save(
+            self, api_v2_client, api_cookies, new_executors_ability_payload
+    ):
+        payload = copy.deepcopy(new_executors_ability_payload)
+        payload['id'] = '../upload test/sanitized-id!'
+        expected_id = 'uploadtestsanitized-id'
+
+        try:
+            resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=payload)
+
+            assert resp.status == HTTPStatus.OK
+            ability_data = await resp.json()
+            assert ability_data['ability_id'] == expected_id
+            assert os.path.exists(f'data/abilities/{payload["tactic"]}/{expected_id}.yml')
+        finally:
+            ability_file_cleanup(payload['tactic'], expected_id)
+
+    async def test_create_ability_without_id_generates_id(
+            self, api_v2_client, api_cookies, new_executors_ability_payload
+    ):
+        payload = copy.deepcopy(new_executors_ability_payload)
+        payload.pop('id')
+        ability_id = None
+
+        try:
+            resp = await api_v2_client.post('/api/v2/abilities', cookies=api_cookies, json=payload)
+
+            assert resp.status == HTTPStatus.OK
+            ability_data = await resp.json()
+            ability_id = ability_data['ability_id']
+            assert ability_id
+            assert os.path.exists(f'data/abilities/{payload["tactic"]}/{ability_id}.yml')
+        finally:
+            if ability_id:
+                ability_file_cleanup(payload['tactic'], ability_id)
+
     async def test_create_ability_accepts_safe_payload_and_plugin_style_executor_names(
             self, api_v2_client, api_cookies, new_executors_ability_payload
     ):

From 5b55546eae06dd4a45063f868db8d00c86d027c2 Mon Sep 17 00:00:00 2001
From: Fiona McCrae <fmccrae@mitre.org>
Date: Wed, 6 May 2026 16:08:52 -0400
Subject: [PATCH 10/10] added validation for executors

---
 app/objects/secondclass/c_executor.py | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/app/objects/secondclass/c_executor.py b/app/objects/secondclass/c_executor.py
index 4a5970ea7..8c425a2cb 100644
--- a/app/objects/secondclass/c_executor.py
+++ b/app/objects/secondclass/c_executor.py
@@ -21,6 +21,18 @@ class ExecutorSchema(ma.Schema):
     variations = ma.fields.List(ma.fields.Nested(VariationSchema()))
     additional_info = ma.fields.Dict(keys=ma.fields.String(), values=ma.fields.String())
 
+    @ma.validates_schema
+    def validate_required_executor_fields(self, data, **kwargs):
+        if kwargs.get('partial') is True:
+            return
+
+        errors = {}
+        for field_name in ('name', 'platform'):
+            if not data.get(field_name):
+                errors[field_name] = ['Missing data for required field.']
+        if errors:
+            raise ma.ValidationError(errors)
+
     @ma.post_load
     def build_executor(self, data, **_):
         return Executor(**data)