Merge pull request #64 from splunk/add_appinspect_rest_support

Add appinspect rest support

contentctl/actions/api_deploy.py

@@ -36,11 +36,11 @@ def fix_newlines_in_conf_files(self, conf_path: pathlib.Path) -> RawConfigParser
     def execute(self, input_dto: API_DeployInputDto) -> None:
         if len(input_dto.config.deployments.rest_api_deployments) == 0:
             raise Exception("No rest_api_deployments defined in 'contentctl.yml'")
-        app_path =  pathlib.Path(input_dto.config.build.path_root)/input_dto.config.build.name
+        app_path =  pathlib.Path(input_dto.config.build.path_root)/input_dto.config.build.title
         if not app_path.is_dir():
             raise Exception(f"The unpackaged app does not exist at the path {app_path}. Please run 'contentctl build' to generate the app.")
         for target in input_dto.config.deployments.rest_api_deployments:
-            print(f"Deploying '{input_dto.config.build.name}' to target '{target.server}' [{target.description}]")
+            print(f"Deploying '{input_dto.config.build.title}' to target '{target.server}' [{target.description}]")
             splunk_args = {
                 "host": target.server,
                 "port": target.port,

contentctl/actions/generate.py

@@ -9,11 +9,13 @@
 from contentctl.output.conf_output import ConfOutput
 from contentctl.output.ba_yml_output import BAYmlOutput
 from contentctl.output.api_json_output import ApiJsonOutput
-
+from typing import Union
 
 @dataclass(frozen=True)
 class GenerateInputDto:
     director_input_dto: DirectorInputDto
+    appinspect_api_username: Union[str,None] = None
+    appinspect_api_password: Union[str,None] = None
 
 
 class Generate:
@@ -24,6 +26,14 @@ def execute(self, input_dto: GenerateInputDto) -> DirectorOutputDto:
         director.execute(input_dto.director_input_dto)
 
         if input_dto.director_input_dto.product == SecurityContentProduct.SPLUNK_APP:
+            if input_dto.appinspect_api_username and input_dto.appinspect_api_password:
+                pass
+            elif input_dto.appinspect_api_username or input_dto.appinspect_api_password:
+                if input_dto.appinspect_api_password:
+                    raise Exception("appinspect_api_password was provided, but appinspect_api_username was not. Please provide both or neither")
+                else:
+                    raise Exception("appinspect_api_username was provided, but appinspect_api_password was not. Please provide both or neither")            
+
             conf_output = ConfOutput(input_dto.director_input_dto.input_path, input_dto.director_input_dto.config)
             conf_output.writeHeaders()
             conf_output.writeObjects(director_output_dto.detections, SecurityContentType.detections)
@@ -34,8 +44,11 @@ def execute(self, input_dto: GenerateInputDto) -> DirectorOutputDto:
             conf_output.writeObjects(director_output_dto.macros, SecurityContentType.macros)
             conf_output.writeAppConf()
             conf_output.packageApp()
-            #conf_output.inspectApp()
 
+            conf_output.inspectAppCLI()
+            if input_dto.appinspect_api_username and input_dto.appinspect_api_password:
+                conf_output.inspectAppAPI(input_dto.appinspect_api_username, input_dto.appinspect_api_password)
+
             print(f'Generate of security content successful to {conf_output.output_path}')
             return director_output_dto
 

contentctl/contentctl.py

@@ -122,7 +122,7 @@ def build(args, config:Union[Config,None]=None) -> DirectorOutputDto:
         product=product_type, 
         config=config
     )
-    generate_input_dto = GenerateInputDto(director_input_dto)
+    generate_input_dto = GenerateInputDto(director_input_dto, args.appinspect_api_username, args.appinspect_api_password)
 
     generate = Generate()
 
@@ -212,11 +212,11 @@ def test(args: argparse.Namespace):
     # be able to do it in Test().execute. For now, we will do it here
     app = App(
         uid=9999,
-        appid=config.build.name,
-        title=config.build.name,
+        appid=config.build.title,
+        title=config.build.title,
         release=config.build.version,
         http_path=None,
-        local_path=str(pathlib.Path(config.build.path_root)/f"{config.build.name}-{config.build.version}.tar.gz"),
+        local_path=str(pathlib.Path(config.build.path_root)/f"{config.build.title}-{config.build.version}.tar.gz"),
         description=config.build.description,
         splunkbase_path=None,
         force_local=True
@@ -424,6 +424,22 @@ def main():
         default="app",
         help="Type of package: app, ssa or api"
     )
+
+    build_parser.add_argument(
+        "--appinspect_api_username",
+        required=False,
+        type=str,
+        default=None,
+        help=f"Username for running AppInspect on {SecurityContentProduct.SPLUNK_APP.name} ONLY. For documentation, please review https://dev.splunk.com/enterprise/reference/appinspect/appinspectapiepref"
+    )
+    build_parser.add_argument(
+        "--appinspect_api_password",
+        required=False,
+        type=str,
+        default=None,
+        help=f"Password for running AppInspect on {SecurityContentProduct.SPLUNK_APP.name} ONLY. For documentation, please review https://dev.splunk.com/enterprise/reference/appinspect/appinspectapiepref"
+    )
+
     build_parser.set_defaults(func=build)
 
     docs_parser.set_defaults(func=doc_gen)
@@ -500,6 +516,24 @@ def main():
                                                  "contentctl_test.yml.")
     test_parser.add_argument("--num_containers", required=False, default=1, type=int)
     test_parser.add_argument("--server_info", required=False, default=None, type=str, nargs='+')
+
+    #Even though these are also options to build, make them available to test_parser
+    #as well to make the tool easier to use
+    test_parser.add_argument(
+        "--appinspect_api_username",
+        required=False,
+        type=str,
+        default=None,
+        help=f"Username for running AppInspect on {SecurityContentProduct.SPLUNK_APP.name} ONLY. For documentation, please review https://dev.splunk.com/enterprise/reference/appinspect/appinspectapiepref"
+    )
+    test_parser.add_argument(
+        "--appinspect_api_password",
+        required=False,
+        type=str,
+        default=None,
+        help=f"Password for running AppInspect on {SecurityContentProduct.SPLUNK_APP.name} ONLY. For documentation, please review https://dev.splunk.com/enterprise/reference/appinspect/appinspectapiepref"
+    )
+
     test_parser.set_defaults(func=test)
 
     convert_parser.add_argument("-dm", "--data_model", required=False, type=str, default="cim", help="converter target, choose between cim, raw, ocsf")

contentctl/objects/abstract_security_content_objects/detection_abstract.py

@@ -84,7 +84,7 @@ def type_valid(cls, v, values):
             raise ValueError("not valid analytics type: " + values["name"])
         return v
 
-    @validator('how_to_implement')
+    @validator('how_to_implement', 'search', 'known_false_positives')
     def encode_error(cls, v, values, field):
         return SecurityContentObject.free_text_field_valid(cls,v,values,field)
 
@@ -112,43 +112,42 @@ def validation_for_ba_only(cls, values):
     #     return v
 
 
-    # @validator("search")
-    # def search_obsersables_exist_validate(cls, v, values):
-    #     if type(v) is str:
-    #         tags:DetectionTags = values.get("tags")
-    #         if tags == None:
-    #             raise ValueError("Unable to parse Detection Tags.  Please resolve Detection Tags errors")
+    @validator("search")
+    def search_obsersables_exist_validate(cls, v, values):
+        if type(v) is str:
+            tags:DetectionTags = values.get("tags")
+            if tags == None:
+                raise ValueError("Unable to parse Detection Tags.  Please resolve Detection Tags errors")
 
-    #         observable_fields = [ob.name.lower() for ob in tags.observable]
+            observable_fields = [ob.name.lower() for ob in tags.observable]
 
-    #         #All $field$ fields from the message must appear in the search
-    #         field_match_regex = r"\$([^\s.]*)\$"
+            #All $field$ fields from the message must appear in the search
+            field_match_regex = r"\$([^\s.]*)\$"
 
-    #         message_fields = [match.replace("$", "").lower() for match in re.findall(field_match_regex, tags.message.lower())]
-    #         missing_fields = set([field for field in observable_fields if field not in v.lower()])
+            message_fields = [match.replace("$", "").lower() for match in re.findall(field_match_regex, tags.message.lower())]
+            missing_fields = set([field for field in observable_fields if field not in v.lower()])
 
-    #         error_messages = []
-    #         if len(missing_fields) > 0:
-    #             error_messages.append(f"The following fields are declared as observables, but do not exist in the search: {missing_fields}")
+            error_messages = []
+            if len(missing_fields) > 0:
+                error_messages.append(f"The following fields are declared as observables, but do not exist in the search: {missing_fields}")
 
 
-    #         missing_fields = set([field for field in message_fields if field not in v.lower()])
-    #         if len(missing_fields) > 0:
-    #             error_messages.append(f"The following fields are used as fields in the message, but do not exist in the search: {missing_fields}")
+            missing_fields = set([field for field in message_fields if field not in v.lower()])
+            if len(missing_fields) > 0:
+                error_messages.append(f"The following fields are used as fields in the message, but do not exist in the search: {missing_fields}")
 
-    #         if len(error_messages) > 0 and values.get("status") == DetectionStatus.production.value:
-    #             msg = "\n\t".join(error_messages)
-    #             print("Errors found in notable validation - skipping for now")
-    #             #raise(ValueError(msg))
+            if len(error_messages) > 0 and values.get("status") == DetectionStatus.production.value:
+                msg = "\n\t".join(error_messages)
+                raise(ValueError(msg))
 
-    #     # Found everything
-    #     return v
+        # Found everything
+        return v
 
     @validator("tests")
     def tests_validate(cls, v, values):
-        if values.get("status","") != DetectionStatus.production and not v:
+        if values.get("status","") == DetectionStatus.production.value and not v:
             raise ValueError(
-                "tests value is needed for production detection: " + values["name"]
+                "One or more tests is REQUIRED for production detection: " + values["name"]
             )
         return v
 

contentctl/objects/abstract_security_content_objects/security_content_object_abstract.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
-
+import re
 import abc
 import string
 import uuid
 from datetime import datetime
 from pydantic import BaseModel, validator, ValidationError, Field
 from contentctl.objects.enums import SecurityContentType
 from typing import Tuple
+
 import uuid
 import pathlib
 
@@ -44,14 +45,21 @@ def date_valid(cls, v, values):
 
     @staticmethod
     def free_text_field_valid(input_cls, v, values, field):
+
         try:
             v.encode('ascii')
-        except UnicodeEncodeError:
-            raise ValueError('encoding error in ' + field.name + ': ' + values["name"])
+        except UnicodeEncodeError as e:
+            print(f"Potential Ascii encoding error in {values['name']}:{field.name} - {str(e)}")
+
+
+        if bool(re.search(r"[^\\]\n", v)):
+                raise ValueError(f"Unexpected newline(s) in {values['name']}:{field.name}.  Newline characters MUST be prefixed with \\")
         return v
 
-    @validator('description')
+
+    @validator("name", "author", 'description')
     def description_valid(cls, v, values, field):
+
         return SecurityContentObject_Abstract.free_text_field_valid(cls,v,values,field)
 
 
@@ -70,4 +78,5 @@ def create_filename_to_content_dict(all_objects:list[SecurityContentObject_Abstr
             name_dict[str(pathlib.Path(object.file_path))] = object
 
         return name_dict
-
+
+

contentctl/objects/config.py

@@ -99,7 +99,7 @@ class ConfigBuildBa(BaseModel):
 class ConfigBuild(BaseModel):
     # Fields required for app.conf based on
     # https://docs.splunk.com/Documentation/Splunk/9.0.4/Admin/Appconf
-    name: str = Field(default="ContentPack",title="Internal name used by your app.  No spaces or special characters.")
+    title: str = Field(default="ContentPack",title="Internal name used by your app.  No spaces or special characters.")
     path_root: str = Field(default="dist",title="The root path at which you will build your app.")
     prefix: str = Field(default="ContentPack",title="A short prefix to easily identify all your content.")
     build: int = Field(default=int(datetime.utcnow().strftime("%Y%m%d%H%M%S")),
@@ -120,7 +120,7 @@ class ConfigBuild(BaseModel):
     # * must not be any of the following names: CON, PRN, AUX, NUL,
     #   COM1, COM2, COM3, COM4, COM5, COM6, COM7, COM8, COM9,
     #   LPT1, LPT2, LPT3, LPT4, LPT5, LPT6, LPT7, LPT8, LPT9
-    id: str = Field(default="ContentPack",title="Internal name used by your app.  No spaces or special characters.")
+    name: str = Field(default="ContentPack",title="Internal name used by your app.  No spaces or special characters.")
     label: str = Field(default="Custom Splunk Content Pack",title="This is the app name that shows in the launcher.")
     author_name: str = Field(default="author name",title="Name of the Content Pack Author.")
     author_email: str = Field(default="[email protected]",title="Contact email for the Content Pack Author")
@@ -138,6 +138,12 @@ def validate_version(cls, v, values):
         except Exception as e:
             raise(ValueError(f"The specified version does not follow the semantic versioning spec (https://semver.org/). {str(e)}"))
         return v
+
+    #Build will ALWAYS be the current utc timestamp
+    @validator('build', always=True)
+    def validate_build(cls, v, values):
+        return int(datetime.utcnow().strftime("%Y%m%d%H%M%S"))
+
 
 
 

contentctl/objects/playbook.py

@@ -29,4 +29,8 @@ class Playbook(SecurityContentObject):
 
     @validator('references')
     def references_check(cls, v, values):
-        return LinkValidator.SecurityContentObject_validate_references(v, values)
+        return LinkValidator.SecurityContentObject_validate_references(v, values)
+
+    @validator('how_to_implement')
+    def encode_error(cls, v, values, field):
+        return SecurityContentObject.free_text_field_valid(cls,v,values,field)