Add host-info and create-user functions

Author: mraible

functions/host-info/main.py

@@ -0,0 +1,32 @@
+from falconfoundry import FoundryFunction, FoundryRequest, FoundryResponse, FoundryAPIError
+# The following two lines are necessary to import utils
+import sys
+sys.path.append('../')
+from utils import validate_host_id, format_error_response
+from logging import Logger
+from typing import Dict
+
+
+func = FoundryFunction.instance()
+
+# Handler on_post
+@func.handler(method='POST', path='/host-info')
+def on_post(request: FoundryRequest, config: Dict[str, object] | None, logger: Logger) -> FoundryResponse:
+    host_id = request.body.get('host_id')
+
+    logger.info(f"Host ID: {host_id}")
+    logger.info(f"Is valid? {validate_host_id(host_id)}")
+
+    if not validate_host_id(host_id):
+        return FoundryResponse(errors=format_error_response("Invalid host ID format"))
+
+    return FoundryResponse(
+        code=200,
+        body={
+            "host":host_id
+        }
+    )
+
+
+if __name__ == '__main__':
+    func.run()

functions/host-info/request_schema.json

@@ -0,0 +1,12 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "properties": {
+    "name": {
+      "type": "string"
+    }
+  },
+  "required": [
+    "host_id"
+  ],
+  "type": "object"
+}

functions/host-info/requirements.txt

@@ -0,0 +1 @@
+crowdstrike-falcon-foundry==1.0.0

functions/host-info/response_schema.json

@@ -0,0 +1,9 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "properties": {
+    "host": {
+      "type": "string"
+    }
+  },
+  "type": "object"
+}

functions/servicenow/main.py

@@ -1,17 +1,13 @@
 from falconfoundry import FoundryFunction, FoundryRequest, FoundryResponse, FoundryAPIError
 from falconpy import APIIntegrations
-import logging
+from logging import Logger
+from typing import Dict
 
 
 func = FoundryFunction.instance()
 
-# This configuration for logging will only work when running locally
-# See the Add Logging section below for how to log to NG-SIEM
-logger = logging.getLogger()
-logger.setLevel(logging.INFO)
-
 @func.handler(method='POST', path='/ticket')
-def on_post(request: FoundryRequest) -> FoundryResponse:
+def on_post(request: FoundryRequest, config: Dict[str, object] | None, logger: Logger) -> FoundryResponse:
     """
     Create an incident ticket in ServiceNow using the Table API.
 

functions/user-management/main.py

@@ -0,0 +1,32 @@
+from falconfoundry import FoundryFunction, FoundryRequest, FoundryResponse, FoundryAPIError
+# The following two lines are necessary to import utils
+import sys
+sys.path.append('../')
+from utils import validate_email, format_error_response
+from logging import Logger
+from typing import Dict
+
+
+func = FoundryFunction.instance()
+
+# Handler on_post
+@func.handler(method='POST', path='/create-user')
+def on_post(request: FoundryRequest, config: Dict[str, object] | None, logger: Logger) -> FoundryResponse:
+    email = request.body.get('email')
+
+    logger.info(f"Email: {email}")
+    logger.info(f"Is valid? {validate_email(email)}")
+
+    if not validate_email(email):
+        return FoundryResponse(errors=format_error_response("Invalid email format"))
+
+    return FoundryResponse(
+        code=200,
+        body={
+            "email":email
+        }
+    )
+
+
+if __name__ == '__main__':
+    func.run()

functions/user-management/request_schema.json

@@ -0,0 +1,12 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "properties": {
+    "email": {
+      "type": "string"
+    }
+  },
+  "required": [
+    "email"
+  ],
+  "type": "object"
+}

functions/user-management/requirements.txt

@@ -0,0 +1 @@
+crowdstrike-falcon-foundry==1.0.0

functions/user-management/response_schema.json

@@ -0,0 +1,9 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "properties": {
+    "email": {
+      "type": "string"
+    }
+  },
+  "type": "object"
+}

functions/utils.py

@@ -0,0 +1,26 @@
+import re
+import json
+from typing import Dict, Any, Optional, Union, List
+from falconfoundry import FoundryAPIError
+
+def validate_host_id(host_id: str) -> bool:
+    """Validate that a host ID is in the correct format."""
+    if not host_id or not isinstance(host_id, str):
+        return False
+    return len(host_id) == 32 and all(c in '0123456789abcdef' for c in host_id.lower())
+
+def validate_email(email: str) -> bool:
+    """Validate email format."""
+    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
+    return bool(re.match(pattern, email))
+
+def format_error_response(message: str, code: int = 400) -> list[Any]:
+    """Create a standardized error response."""
+    return [FoundryAPIError(code=code, message=message)]
+
+def safe_json_parse(data: str) -> Optional[Dict[str, Any]]:
+    """Safely parse JSON data."""
+    try:
+        return json.loads(data)
+    except (json.JSONDecodeError, TypeError):
+        return None