initial commit

Author: mowsec

README.md

@@ -1,2 +1,17 @@
-# contrast-integrations-cli
+# The Contrast Security Integrations CLI
 A CLI tool for adding Contrast Integrations via rule customizations.
+
+Adapted from [Contrast-Security-OSS/integrations-scw](https://github.com/Contrast-Security-OSS/integrations-scw) to include a CLI tool, and also to include Secure Flag, Secure Code Warrior and possibly other integrations in the future.
+
+## Usage
+```sh
+contrast-integrations --help
+contrast-integrations auth init
+contrast-integrations secure-code-warrior enable-for-rule sql-injection
+contrast-integrations secure-code-warrior disable-for-rule sql-injection
+
+contrast-integrations secure-flag enable-for-all 
+contrast-integrations secure-flag disable-for-all
+
+
+```

contrast_api/api_handlers.py

@@ -0,0 +1,103 @@
+import json
+import requests
+
+from rich import print
+
+from contrast_api.models import *
+from contrast_api.request_handler import RequestHandler
+
+class BaseAPI:
+    api_version = "api/ng"
+
+    def __init__(self, session, api_config):
+        self._session = session
+        self._config = api_config
+
+        self.hostname = self._config.teamserver_url
+        self.requests = RequestHandler(session=self._session)
+
+    @property
+    def _base_url(self) -> str:
+        base_url = self.hostname 
+        base_url += f"/{self.api_version}" if self.api_version else ""
+        return base_url
+    
+    @property
+    def current_org_uuid(self):
+        if self._config.active_profile:
+            return self._config.active_profile.org_uuid
+        else:
+            return self._config.default_profile.org_uuid
+    
+    def full_url(self, endpoint: str) -> str:
+        return f"{self._base_url}/{endpoint}"
+    
+    def get(self, endpoint: str, skip_links: bool = False, expand: str = None, params: Dict = None) -> Response:
+        if any([skip_links, expand, params]):
+            params = params or {}
+            expand = expand or ""
+            if skip_links:
+                expand = "skip_links," + expand
+            if expand: 
+                params["expand"] = expand
+        return self.requests.get(url=self.full_url(endpoint), params=params)
+
+    def post(self, endpoint: str, skip_links: bool = False, expand: List = None, params: Dict = None, data: Dict = {}) -> Response:
+        if any([skip_links, expand, params]):
+            params = params or {}
+            expand = expand or ""
+            if skip_links:
+                expand = "skip_links," + expand
+            if expand: 
+                params["expand"] = expand
+        return self.requests.post(url=self.full_url(endpoint), params=params, data=data)
+
+
+class APIHandler(BaseAPI):
+    api_version = ""
+
+    def get_current_user_profile(self):
+        print("Getting current user profile")
+        return self.get("profile/current-user")
+
+
+class PolicyHandler(BaseAPI):
+
+    def get_org_policy(self) -> Policy:
+        response = self.get(f"{self.current_org_uuid}/rules", skip_links=True)
+        return Policy.from_response(response)
+    
+    def get_details_for_rule(self, rule_name: str) -> Rule:
+        response = self.get(f"{self.current_org_uuid}/rules/{rule_name}")
+        return Rule.from_response(response, _policy_handler=self)
+    
+    def reset_rule(self, rule_name: str) -> Response:
+        response = self.post(f"{self.current_org_uuid}/rules/{rule_name}", data={"override": "false"})
+        return response
+    
+    def update_rule(self, rule_name: str, customizations: Dict) -> Response:
+        response = self.post(f"{self.current_org_uuid}/rules/{rule_name}", data=customizations)
+        return response
+
+
+class ProfileHandler(BaseAPI):
+
+    def get_profile(self):
+        response = self.get("profile", expand="email,preferences,login,signup,service_key,ip_address")
+        return User.from_response(response)
+
+    def get_current_user_profile(self):
+        response = self.get("profile/current-user", expand="ip_address")
+        return User.from_response(response)
+
+    def get_organizations_in_profile(self):
+        response = self.get("profile/organizations", expand="role")
+        return Organization.from_response(response)
+    
+    def get_default_org_from_profile(self):
+        response = self.get("profile/organizations/default")
+        return Organization.from_response(response)
+    
+    def get_profile_user_roles_for_org(self):
+        response = self.get(f"profile/organizations/{self.current_org_uuid}", expand="role")
+        return Organization.from_response(response)

contrast_api/contrast_api.py

@@ -0,0 +1,107 @@
+import json
+import logging
+import requests
+import yaml 
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, List, Optional
+from rich import print
+from rich.logging import RichHandler
+from uuid import UUID
+
+from contrast_api.api_handlers import * 
+
+logging.basicConfig(level="INFO", datefmt="[%X]", format="%(name)s: %(message)s", handlers=[RichHandler(markup=True)])
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ProfileAuth(yaml.YAMLObject):
+    org_uuid: UUID
+    api_key: str
+    name: Optional[str] = None
+
+    @classmethod
+    def from_dict(cls):
+        pass
+
+
+@dataclass
+class ContrastAPIConfig(yaml.YAMLObject):
+    teamserver_url: str
+    api_key: str
+    auth_header: str
+    is_superadmin: Optional[bool] = False
+    default_profile: Optional[ProfileAuth] = None
+    active_profile: Optional[ProfileAuth] = None
+
+    @classmethod
+    def from_yaml_file(cls, config_file: Path):
+        """Load ContrastAuth object from a YAML file"""
+        try: 
+            logger.debug(f"Loading config file from {config_file}")
+            contrast_config = yaml.safe_load(config_file.read_text())
+            return cls(**contrast_config)
+        except FileNotFoundError as err:
+            logger.error(f"Could not find config file!: {err}")
+            raise err
+    
+    def __post_init__(self):
+        if self.default_profile:
+            self.default_profile = ProfileAuth(**self.default_profile)
+        if self.active_profile:
+            self.active_profile = ProfileAuth(**self.active_profile)
+        
+    def to_yaml_file(self, config_file: Path):
+        """Write ContrastAuth object to a YAML file"""
+        logger.debug(f"Writing config file to '{config_file}'")
+        with open(config_file, "w") as file:
+            yaml.emitter.Emitter.process_tag = lambda self, *args, **kw: None
+            yaml.dump(self.__dict__, file, sort_keys=False, allow_unicode=True, default_flow_style=False)
+    
+    def set_default_profile(self, org_uuid: str, api_key: str, config_file: Path, org_name: str = None):
+        self.default_profile = ProfileAuth(name=org_name, org_uuid=org_uuid, api_key=api_key)
+        self.to_yaml_file(config_file)
+
+    def set_active_profile(self, org_uuid: str, api_key: str, config_file: Path, org_name: str = None):
+        self.active_profile = ProfileAuth(name=org_name, org_uuid=org_uuid, api_key=api_key)
+        self.to_yaml_file(config_file)
+            
+
+@dataclass
+class ContrastAPIResponse:
+    status_code: int
+    message: str
+    data: List[Dict]
+
+
+class ContrastAPI:
+
+    def __init__(self, session: requests.Session, api_config: ContrastAPIConfig):
+        self.api_config = api_config
+        self.session = session
+        self.api_config = api_config
+
+    @classmethod
+    def from_config(cls, api_config: ContrastAPIConfig):
+        session = requests.Session()
+        session.headers.update({
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+            "API-Key": api_config.active_profile.api_key if api_config.active_profile else api_config.api_key,
+            "Authorization": api_config.auth_header
+        })
+        return cls(session, api_config)
+    
+    @property
+    def base(self):
+        return BaseAPI(self.session, self.api_config)
+    
+    @property
+    def policy(self):
+        return PolicyHandler(self.session, self.api_config)
+
+    @property
+    def profile(self):
+        return ProfileHandler(self.session, self.api_config)