Sample Agentic AI Orchestrating Complex Cybersecurity Workflow in GCP

Estimated reading time: 7 minutes

Agentic AI Orchestrating Complex Workflow in GCP (Sample)

This sample outlines a conceptual implementation of an agentic AI system orchestrating a complex cybersecurity workflow in Google Cloud Platform (GCP), focusing on automatically investigating and responding to potential phishing incidents reported by employees.

Conceptual Architecture

        
+---------------------+     +---------------------+     +--------------------+     +---------------------+     +---------------------+     +------------------+
| Employee Phishing   | --> | Cloud Pub/Sub Topic | --> | Agentic AI Workflow| --> | Google Cloud        | --> | Security Team       | --> | Further Automated   |
| Reports (e.g., Gmail|     | (Phishing Reports)  |     | (Cloud Functions / |     | Services (Virus     |     | Notifications       |     | Actions (e.g., User|
| Add-on, SIEM Alert)|     |                     |     | Vertex AI Agents)   |     | Total, Chronicle,   |     | (Email, Slack,      |     | Password Reset,     |
+---------------------+     +---------------------+     +--------------------+     | Security Command    |     | etc.)               |     | Account Suspension)|
                                                                                   | Center)             |     +---------------------+     +-------------------+
                                                                                   +---------------------+

Explanation of Components

Employee Phishing Reports: Employees report suspicious emails through a designated channel.
Cloud Pub/Sub Topic (Phishing Reports): A messaging service decoupling the reporting mechanism from the workflow.
Agentic AI Workflow (Cloud Functions / Vertex AI Agents): The core of the agentic system, responsible for investigation and response orchestration.
- Google Cloud Functions: Serverless functions acting as orchestrators and specialized task handlers.
- Vertex AI Agents (Preview): A platform for building conversational AI agents that can interact with GCP services.
Google Cloud Services: Tools used for information gathering and action taking.
- VirusTotal: Analyzes email attachments, URLs, and sender information.
- Chronicle: Google’s cloud-native SIEM for security analytics.
- Security Command Center (SCC): Provides a central view of security risks.
Security Team Notifications: Alerts sent to the security team based on investigation results.
Further Automated Actions: Actions triggered based on the threat assessment (e.g., password reset, account suspension).

Conceptual Implementation Steps

Set up Cloud Pub/Sub: Create a Pub/Sub topic named phishing-reports.

Implement the Agentic AI Workflow (using Cloud Functions as the orchestrator):

Orchestrator Function (Python Example – Conceptual):

                        
import json
import os
from google.cloud import pubsub_v1
from google.cloud import functions_framework

PROJECT_ID = os.environ.get("GCP_PROJECT")
INVESTIGATE_FUNCTION = f"https://{os.environ.get('FUNCTION_REGION')}-{PROJECT_ID}.cloudfunctions.net/investigate-phishing"
RESPOND_FUNCTION = f"https://{os.environ.get('FUNCTION_REGION')}-{PROJECT_ID}.cloudfunctions.net/respond-to-phishing"
NOTIFICATION_FUNCTION = f"https://{os.environ.get('FUNCTION_REGION')}-{PROJECT_ID}.cloudfunctions.net/notify-security-team"

@functions_framework.cloud_event
def orchestrate_phishing_workflow(cloud_event):
    data = json.loads(cloud_event.data.decode('utf-8'))
    report_details = data.get('report')

    print(f"Received phishing report: {report_details}")

    # Initial investigation - trigger the investigation function
    client = pubsub_v1.PublisherClient()
    topic_path = client.topic_path(PROJECT_ID, 'investigation-tasks')
    message_data = json.dumps({'report': report_details}).encode('utf-8')
    future = client.publish(topic_path, message_data)
    print(f"Published investigation task: {future.result()}")

    # In a more sophisticated system, the orchestrator would track the investigation
    # progress and trigger subsequent steps based on the findings.
    # For this simple example, we'll just trigger the initial investigation.

Investigation Function (Python Example – Conceptual):

                        
import json
import os
import requests
from google.cloud import pubsub_v1
from google.cloud import functions_framework

PROJECT_ID = os.environ.get("GCP_PROJECT")
VT_API_KEY = os.environ.get("VIRUSTOTAL_API_KEY")
CHRONICLE_API_ENDPOINT = os.environ.get("CHRONICLE_API_ENDPOINT")
RESPOND_TOPIC = 'respond-tasks'
NOTIFICATION_TOPIC = 'notification-tasks'

@functions_framework.cloud_event
def investigate_phishing(cloud_event):
    data = json.loads(cloud_event.data.decode('utf-8'))
    report = data.get('report')
    email_details = report.get('email')

    print(f"Investigating phishing report for: {email_details.get('sender')}")

    investigation_results = {}

    # --- Example: Analyze Sender with VirusTotal ---
    if VT_API_KEY and email_details.get('sender'):
        vt_url = f"https://www.virustotal.com/api/v3/domains/{email_details['sender'].split('@')[1]}"
        headers = {"x-apikey": VT_API_KEY}
        response = requests.get(vt_url, headers=headers)
        if response.status_code == 200:
            investigation_results['virustotal_sender'] = response.json()
        else:
            print(f"Error querying VirusTotal: {response.status_code}")

    # --- Example: Query Chronicle for similar emails (Conceptual) ---
    if CHRONICLE_API_ENDPOINT and email_details.get('subject'):
        # Implement logic to query Chronicle API
        investigation_results['chronicle_results'] = "Chronicle query performed (results not shown)"

    # --- Decide on next steps based on investigation results ---
    risk_score = calculate_risk_score(investigation_results)
    print(f"Calculated risk score: {risk_score}")

    client = pubsub_v1.PublisherClient()

    if risk_score > 0.7:
        # Trigger response function
        topic_path = client.topic_path(PROJECT_ID, RESPOND_TOPIC)
        message_data = json.dumps({'report': report, 'investigation': investigation_results, 'risk_score': risk_score}).encode('utf-8')
        future = client.publish(topic_path, message_data)
        print(f"Published response task: {future.result()}")
    elif risk_score > 0.3:
        # Notify security team for further review
        topic_path = client.topic_path(PROJECT_ID, NOTIFICATION_TOPIC)
        message_data = json.dumps({'report': report, 'investigation': investigation_results, 'risk_score': risk_score}).encode('utf-8')
        future = client.publish(topic_path, message_data)
        print(f"Published notification task: {future.result()}")
    else:
        print("Phishing report assessed as low risk.")

def calculate_risk_score(results):
    score = 0
    if results.get('virustotal_sender') and results['virustotal_sender'].get('data') and results['virustotal_sender']['data'].get('attributes'):
        attributes = results['virustotal_sender']['data']['attributes']
        if attributes.get('reputation')  0:
            score += 0.3
    # Add more scoring logic based on Chronicle results, etc.
    return score

Response Function (Python Example – Conceptual):

                        
import json
import os
from google.cloud import functions_framework
# Import GCP client libraries for actions like password reset, etc.

@functions_framework.cloud_event
def respond_to_phishing(cloud_event):
    data = json.loads(cloud_event.data.decode('utf-8'))
    report = data.get('report')
    investigation = data.get('investigation')
    risk_score = data.get('risk_score')

    print(f"Responding to high-risk phishing report for: {report.get('email').get('recipient')}")
    print(f"Investigation results: {investigation}")

    # --- Example: Trigger user password reset (Conceptual) ---
    # Implement interaction with Cloud Identity API

    # --- Example: Quarantine email (Conceptual) ---
    # Implement interaction with Gmail API

    # --- Notify security team about automated actions ---
    # Trigger the notification function

Notification Function (Python Example – Conceptual):

                        
import json
import os
import requests
from google.cloud import functions_framework

SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL")

@functions_framework.cloud_event
def notify_security_team(cloud_event):
    data = json.loads(cloud_event.data.decode('utf-8'))
    report = data.get('report')
    investigation = data.get('investigation')
    risk_score = data.get('risk_score')

    message = f"Potential phishing incident reported by: {report.get('reporter')}\\n"
    message += f"Subject: {report.get('email').get('subject')}\\n"
    message += f"Risk Score: {risk_score}\\n"
    message += f"Investigation Results: {json.dumps(investigation, indent=2)}"

    if SLACK_WEBHOOK_URL:
        payload = {'text': message}
        response = requests.post(SLACK_WEBHOOK_URL, json=payload)
        if response.status_code == 200:
            print("Notification sent to Slack.")
        else:
            print(f"Error sending notification to Slack: {response.status_code}")
    else:
        print("Slack webhook URL not configured.")

Set up Pub/Sub Subscriptions: Create subscriptions for each Cloud Function to the relevant Pub/Sub topics.
Configure IAM Permissions: Grant the Cloud Functions the necessary permissions to interact with other GCP services and external APIs.
Deploy Cloud Functions: Deploy the orchestrator, investigation, response, and notification Cloud Functions.
Integrate Reporting Mechanism: Configure your phishing reporting system to publish to the phishing-reports topic.

Using Vertex AI Agents (Alternative)

Instead of Cloud Functions, a Vertex AI Agent could be designed to handle the workflow using tools for interacting with security services and a defined conversation flow.

Important Considerations

Complexity of Agentic Logic: The decision-making process of the agent requires careful design.
Error Handling and Logging: Implement robust error handling using Google Cloud Logging.
Security of the Agentic System: Secure API keys and Cloud Function permissions.
Scalability and Performance: Ensure the system can handle the expected load.
Human-in-the-Loop: Integrate human review for critical cases.
Monitoring and Alerting: Monitor the health and performance of the workflow.
Continuous Improvement: Regularly evaluate and refine the agentic logic.

This outline provides a foundation for building an agentic AI-powered phishing response system in GCP. Adapt the implementation based on your specific requirements and environment.

Latest Posts