Refine Operational Security role with redaction and better validation

- Implemented credential redaction in `CloudSecurityAI` to avoid exposing full secrets.
- Added robust error handling and type validation to Flask security endpoints.
- Refactored `OfficialAssistance.jsx` to move tool configurations into a declarative structure.
- Replaced realistic-looking simulated credentials with obviously fake ones.
- Updated unit tests to verify redaction logic.

Co-authored-by: GYFX35 <134739293+GYFX35@users.noreply.github.com>

Author: google-labs-jules[bot]

social_media_analyzer/operational_security.py

@@ -5,12 +5,19 @@
 class CloudSecurityAI:
     """AI for scanning cloud credentials and sensitive information."""
 
+    def _redact(self, value):
+        """Redacts a sensitive string, keeping only the first 4 and last 4 characters."""
+        if len(value) <= 10:
+            return "****"
+        return f"{value[:4]}...{value[-4:]}"
+
     def scan_content(self, text_content):
         findings = {}
         for pattern_name, regex in SENSITIVE_DATA_PATTERNS.items():
             matches = regex.findall(text_content)
             if matches:
-                findings[pattern_name] = matches
+                # Redact each match to avoid full exposure
+                findings[pattern_name] = [self._redact(m) for m in matches]
         return findings
 
 class IoTSecurityAI:

social_media_analyzer/test_operational_security.py

@@ -7,7 +7,8 @@ def test_cloud_security_scan(self):
         content = "My AWS Key is AKIA1234567890ABCDEF"
         findings = ai.scan_content(content)
         self.assertIn("AWS Access Key ID", findings)
-        self.assertEqual(findings["AWS Access Key ID"], ["AKIA1234567890ABCDEF"])
+        # Verify redaction: AKIA1234567890ABCDEF -> AKIA...CDEF
+        self.assertEqual(findings["AWS Access Key ID"], ["AKIA...CDEF"])
 
     def test_iot_security_analyze(self):
         ai = IoTSecurityAI()

src/OfficialAssistance.jsx

@@ -36,9 +36,30 @@ const assistanceRoles = {
     icon: '🔐',
     description: 'Cloud, IoT, and AI-driven security operations for modern infrastructure.',
     tools: [
-      { id: 'cloud_guard', name: 'Cloud Guard', icon: '☁️', desc: 'AI scanner for leaked credentials and sensitive cloud data.' },
-      { id: 'iot_shield', name: 'IoT Shield', icon: '🌐', desc: 'Real-time anomaly detection for industrial IoT networks.' },
-      { id: 'opsec_analyzer', name: 'OpSec Analyzer', icon: '🕵️', desc: 'AI-driven analysis of operational logs for procedural threats.' }
+      {
+        id: 'cloud_guard',
+        name: 'Cloud Guard',
+        icon: '☁️',
+        desc: 'AI scanner for leaked credentials and sensitive cloud data.',
+        endpoint: '/analyze/cloud',
+        getPayload: () => ({ content: "Cloud scan simulation with fake AWS key: AKIA0000000000000000 and fake Google API Key: AIza00000000000000000000000000000000000" })
+      },
+      {
+        id: 'iot_shield',
+        name: 'IoT Shield',
+        icon: '🌐',
+        desc: 'Real-time anomaly detection for industrial IoT networks.',
+        endpoint: '/analyze/iot',
+        getPayload: () => ({ device_data: { voltage: 2.6, temperature: 82, rssi: -95 } })
+      },
+      {
+        id: 'opsec_analyzer',
+        name: 'OpSec Analyzer',
+        icon: '🕵️',
+        desc: 'AI-driven analysis of operational logs for procedural threats.',
+        endpoint: '/analyze/opsec',
+        getPayload: () => ({ logs: ["unauthorized access attempt", "nmap scan detected", "large outbound transfer", "sudo usage"] })
+      }
     ]
   }
 };
@@ -48,38 +69,24 @@ export default function OfficialAssistance() {
   const [analysisResult, setAnalysisResult] = useState(null);
   const [loading, setLoading] = useState(false);
 
-  const handleLaunch = async (toolId, toolName) => {
-    if (!['cloud_guard', 'iot_shield', 'opsec_analyzer'].includes(toolId)) {
-      alert(`Launching ${toolName}... (Simulation mode)`);
+  const handleLaunch = async (tool) => {
+    if (!tool.endpoint) {
+      alert(`Launching ${tool.name}... (Simulation mode)`);
       return;
     }
 
     setLoading(true);
     setAnalysisResult(null);
 
     try {
-      let endpoint = '';
-      let body = {};
-
-      if (toolId === 'cloud_guard') {
-        endpoint = '/analyze/cloud';
-        body = { content: "Sample content with simulated AWS key: AKIA1234567890ABCDEF and Google API Key: AIzaSyA12345678901234567890123456789012" };
-      } else if (toolId === 'iot_shield') {
-        endpoint = '/analyze/iot';
-        body = { device_data: { voltage: 2.6, temperature: 82, rssi: -95 } };
-      } else if (toolId === 'opsec_analyzer') {
-        endpoint = '/analyze/opsec';
-        body = { logs: ["unauthorized access attempt", "nmap scan detected", "large outbound transfer", "sudo usage"] };
-      }
-
-      const response = await fetch(endpoint, {
+      const response = await fetch(tool.endpoint, {
         method: 'POST',
         headers: { 'Content-Type': 'application/json' },
-        body: JSON.stringify(body)
+        body: JSON.stringify(tool.getPayload())
       });
 
       const data = await response.json();
-      setAnalysisResult({ title: toolName, data });
+      setAnalysisResult({ title: tool.name, data });
     } catch (error) {
       console.error("Error launching tool:", error);
       alert("Failed to connect to security backend. Make sure the Flask server is running.");
@@ -117,7 +124,7 @@ export default function OfficialAssistance() {
               </div>
               <button
                 className="action-btn"
-                onClick={() => handleLaunch(tool.id, tool.name)}
+                onClick={() => handleLaunch(tool)}
                 disabled={loading}
               >
                 {loading ? 'Processing...' : 'Launch'}

text_message_analyzer/app.py

@@ -64,8 +64,11 @@ def analyze_cloud():
         return jsonify({"error": "Missing 'content' in request body"}), 400
 
     content = data['content']
-    result = operational_security.analyze_cloud_security(content)
-    return jsonify(result)
+    try:
+        result = operational_security.analyze_cloud_security(content)
+        return jsonify(result)
+    except Exception as e:
+        return jsonify({"error": f"Failed to analyze cloud security: {str(e)}"}), 500
 
 @app.route('/analyze/iot', methods=['POST'])
 def analyze_iot():
@@ -74,8 +77,11 @@ def analyze_iot():
         return jsonify({"error": "Missing 'device_data' in request body"}), 400
 
     device_data = data['device_data']
-    result = operational_security.analyze_iot_security(device_data)
-    return jsonify(result)
+    try:
+        result = operational_security.analyze_iot_security(device_data)
+        return jsonify(result)
+    except Exception as e:
+        return jsonify({"error": f"Failed to analyze IoT security: {str(e)}"}), 500
 
 @app.route('/analyze/opsec', methods=['POST'])
 def analyze_opsec():
@@ -84,8 +90,14 @@ def analyze_opsec():
         return jsonify({"error": "Missing 'logs' in request body"}), 400
 
     logs = data['logs']
-    result = operational_security.analyze_opsec_security(logs)
-    return jsonify(result)
+    if not isinstance(logs, list) or not all(isinstance(log, str) for log in logs):
+        return jsonify({"error": "'logs' must be a list of strings"}), 400
+
+    try:
+        result = operational_security.analyze_opsec_security(logs)
+        return jsonify(result)
+    except Exception as e:
+        return jsonify({"error": f"Failed to analyze operational security: {str(e)}"}), 500
 
 
 if __name__ == '__main__':