Data Leak Detector
by @tobewin
数据泄露检测工具。Use when user wants to scan skills, files, or folders for potential data leaks, privacy risks, or suspicious behavior. Detects network calls, file a...
clawhub install data-leak-detector📖 About This Skill
name: data-leak-detector description: 数据泄露检测工具。Use when user wants to scan skills, files, or folders for potential data leaks, privacy risks, or suspicious behavior. Detects network calls, file access, process spawning, and environment variable access. 数据安全、隐私检测、安全审计。 version: 1.0.0 license: MIT-0 metadata: {"openclaw": {"emoji": "🔍", "requires": {"bins": ["python3"], "env": []}}} dependencies: "pip install watchdog"
Data Leak Detector
Scan skills, files, and folders for potential data leaks and privacy risks.
Features
Risk Levels
| Level | Color | Meaning | |-------|-------|---------| | 🟢 Low | Green | Safe, no concerns | | 🟡 Medium | Yellow | Review recommended | | 🔴 High | Red | Caution required |
Detection Patterns
Network Risks
File Risks
Process Risks
Environment Risks
Trigger Conditions
Python Code
import os
import re
import json
from pathlib import Pathclass DataLeakDetector:
def __init__(self):
self.patterns = {
'network': {
'high': [
r'curl\s+',
r'wget\s+',
r'requests\.(get|post|put|delete)',
r'http[s]?://',
r'urllib\.request',
r'httpx\.',
r'websocket',
],
'medium': [
r'fetch\(',
r'axios\.',
]
},
'file_access': {
'high': [
r'open\s*\(',
r'os\.remove',
r'os\.rmdir',
r'shutil\.rmtree',
],
'medium': [
r'readFile',
r'writeFile',
r'os\.path\.exists',
r'glob\.',
]
},
'process': {
'high': [
r'subprocess\.',
r'os\.system',
r'os\.popen',
r'exec\(',
r'eval\(',
],
'medium': [
r'Popen',
r'call\(',
]
},
'env_access': {
'high': [
r'os\.environ',
r'os\.getenv',
r'\$[A-Z_]+',
],
'medium': [
r'config\[',
r'secrets\[',
]
}
}
def scan_file(self, filepath):
"""Scan a single file for risks"""
risks = []
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
except:
return risks
for category, levels in self.patterns.items():
for level, patterns in levels.items():
for pattern in patterns:
matches = re.finditer(pattern, content, re.IGNORECASE)
for match in matches:
line_num = content[:match.start()].count('\n') + 1
risks.append({
'category': category,
'level': level,
'pattern': pattern,
'line': line_num,
'match': match.group()[:50]
})
return risks
def scan_skill(self, skill_path):
"""Scan entire skill for risks"""
skill_path = Path(skill_path)
all_risks = []
files_scanned = 0
for ext in ['.md', '.py', '.js', '.ts']:
for filepath in skill_path.rglob(f'*{ext}'):
risks = self.scan_file(str(filepath))
for risk in risks:
risk['file'] = str(filepath.relative_to(skill_path))
all_risks.extend(risks)
files_scanned += 1
return all_risks, files_scanned
def calculate_risk_score(self, risks):
"""Calculate overall risk score (0-100)"""
if not risks:
return 0
score = 0
for risk in risks:
if risk['level'] == 'high':
score += 20
elif risk['level'] == 'medium':
score += 10
return min(score, 100)
def generate_report(self, skill_path, risks, files_scanned):
"""Generate risk assessment report"""
risk_score = self.calculate_risk_score(risks)
if risk_score <= 20:
risk_level = "🟢 LOW"
recommendation = "Safe to use"
elif risk_score <= 50:
risk_level = "🟡 MEDIUM"
recommendation = "Review before installing"
else:
risk_level = "🔴 HIGH"
recommendation = "Caution required"
# Group by category
by_category = {}
for risk in risks:
cat = risk['category']
if cat not in by_category:
by_category[cat] = []
by_category[cat].append(risk)
report = []
report.append(f"{'='*60}")
report.append(f"DATA LEAK DETECTOR - SECURITY REPORT")
report.append(f"{'='*60}")
report.append(f"")
report.append(f"Skill: {os.path.basename(skill_path)}")
report.append(f"Files Scanned: {files_scanned}")
report.append(f"Total Risks Found: {len(risks)}")
report.append(f"")
report.append(f"RISK SCORE: {risk_score}/100 ({risk_level})")
report.append(f"RECOMMENDATION: {recommendation}")
report.append(f"")
# Category breakdown
report.append(f"{'='*60}")
report.append(f"RISK BREAKDOWN")
report.append(f"{'='*60}")
for category, category_risks in by_category.items():
high = len([r for r in category_risks if r['level'] == 'high'])
medium = len([r for r in category_risks if r['level'] == 'medium'])
report.append(f"")
report.append(f"{category.upper()}:")
report.append(f" High: {high} | Medium: {medium}")
for risk in category_risks[:3]: # Show top 3
report.append(f" - [{risk['level'].upper()}] {risk['match']} (line {risk['line']})")
# Recommendations
report.append(f"")
report.append(f"{'='*60}")
report.append(f"RECOMMENDATIONS")
report.append(f"{'='*60}")
if 'network' in by_category:
report.append(f"- Review network calls: verify destinations")
if 'file_access' in by_category:
report.append(f"- Review file access: check for sensitive files")
if 'process' in by_category:
report.append(f"- Review subprocess calls: verify commands")
if 'env_access' in by_category:
report.append(f"- Review env access: check for credential access")
return '\n'.join(report)
Example usage
detector = DataLeakDetector()Scan skill
risks, files_scanned = detector.scan_skill('/path/to/skill')
report = detector.generate_report('/path/to/skill', risks, files_scanned)
print(report)
Usage Examples
User: "检查这个skill安全吗"
Agent: Scan SKILL.md and generate risk reportUser: "扫描我的skills有没有数据泄露"
Agent: Scan all installed skills
User: "这个skill有没有网络访问"
Agent: Focus on network risks