文件内容
github_reader_v3_secure.py
"""
GitHub Reader Skill v3.2 — 纯 GitHub API 安全版
相比 v3.1 的变更：
- ❌ 移除所有 zread.ai 第三方依赖（代码、输出、URL）
- ❌ 移除 GitView 本地服务引用
- ✅ 纯 GitHub REST API + 本地智能分析
- ✅ 收紧触发词，避免误触发
- ✅ 安全声明有代码对应，不写空话
- ✅ 新增数据流向透明度手册
安全防护（代码实现对应 SECURITY.md 声明）：
- 输入验证：validate_repo_name() — 正则 + 长度 + 路径遍历检测
- SSRF 防护：safe_url_join() — urllib.parse.quote 编码
- 缓存防投毒：SecureGitHubReaderCache — 大小限制 + JSON 结构验证 + 原子写入
- 路径防遍历：safe_file_path() — os.path.abspath + normpath + startswith 检查
- 并发限制：asyncio.Semaphore
- 速率限制：time-based limiter
- 超时控制：asyncio.wait_for on async calls
"""
import re
import json
import hashlib
import os
import time
import asyncio
import logging
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
from urllib.parse import quote
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)
# ============== 安全配置 ==============
class SecurityConfig:
"""安全配置 - 从环境变量读取"""
# 缓存配置
CACHE_DIR = os.getenv('GITVIEW_CACHE_DIR', '/tmp/gitview_cache')
CACHE_TTL_HOURS = int(os.getenv('GITVIEW_CACHE_TTL', '24'))
CACHE_MAX_SIZE_MB = int(os.getenv('GITVIEW_CACHE_MAX_SIZE', '1'))
# 速率限制
GITHUB_API_DELAY = float(os.getenv('GITVIEW_GITHUB_DELAY', '1.0'))
MAX_CONCURRENT_REQUESTS = int(os.getenv('GITVIEW_MAX_BROWSER', '3'))
# 超时控制
FETCH_TIMEOUT = int(os.getenv('GITVIEW_BROWSER_TIMEOUT', '30'))
GITHUB_API_TIMEOUT = int(os.getenv('GITVIEW_GITHUB_TIMEOUT', '10'))
# 输入验证
MAX_NAME_LENGTH = 100
ALLOWED_NAME_PATTERN = re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9._-]{0,99}$')
# 隐私：数据流向声明
PRIVACY_NOTICE = (
"🔒 **数据流向声明**：本次分析仅使用 GitHub REST API（{api_url}），"
"不会将仓库信息发送给任何第三方服务。"
"分析结果本地缓存 {cache_ttl} 小时，缓存目录：{cache_dir}。"
)
# ============== 安全工具函数 ==============
def validate_repo_name(name: str) -> bool:
"""
验证仓库/所有者名称合法性
安全规则：
1. 只允许字母、数字、-、_、.
2. 必须以字母或数字开头
3. 长度 1-100 字符
4. 禁止 .. 模式（防止路径遍历）
5. 禁止以 - 开头（防止命令行注入）
"""
if not name or not isinstance(name, str):
return False
if len(name) > SecurityConfig.MAX_NAME_LENGTH:
return False
if not SecurityConfig.ALLOWED_NAME_PATTERN.match(name):
return False
if '..' in name:
return False
return True
def safe_url_join(base: str, *paths: str) -> str:
"""安全的 URL 拼接 — urllib.parse.quote 编码路径组件，防 SSRF"""
encoded_paths = [quote(path, safe='') for path in paths]
return '/'.join([base.rstrip('/')] + encoded_paths)
def safe_file_path(base_dir: str, filename: str) -> str:
"""安全的文件路径生成 — 防路径遍历"""
safe_name = re.sub(r'[^a-zA-Z0-9._-]', '', filename)
base_dir = os.path.abspath(base_dir)
file_path = os.path.normpath(os.path.join(base_dir, safe_name))
if not file_path.startswith(base_dir):
raise ValueError(f"Invalid file path: {filename}")
return file_path
# ============== 安全缓存系统 ==============
class SecureGitHubReaderCache:
"""安全的文件缓存系统"""
def __init__(self, cache_dir: str = None):
self.cache_dir = cache_dir or SecurityConfig.CACHE_DIR
self.cache_ttl = timedelta(hours=SecurityConfig.CACHE_TTL_HOURS)
self.max_cache_size = SecurityConfig.CACHE_MAX_SIZE_MB * 1024 * 1024
self._ensure_cache_dir()
def _ensure_cache_dir(self):
try:
os.makedirs(self.cache_dir, exist_ok=True)
os.chmod(self.cache_dir, 0o700)
except Exception as e:
logger.error(f"Failed to create cache directory: {e}")
raise
def _get_cache_key(self, owner: str, repo: str) -> str:
return hashlib.sha256(f"{owner}/{repo}".encode()).hexdigest()
def _get_cache_path(self, key: str) -> str:
return safe_file_path(self.cache_dir, f"{key}.json")
def get(self, owner: str, repo: str) -> Optional[Dict]:
if not validate_repo_name(owner) or not validate_repo_name(repo):
logger.warning(f"Invalid repo name in cache get: {owner}/{repo}")
return None
try:
key = self._get_cache_key(owner, repo)
path = self._get_cache_path(key)
if not os.path.exists(path):
return None
file_size = os.path.getsize(path)
if file_size > self.max_cache_size:
logger.warning(f"Cache file too large: {file_size} bytes")
os.remove(path)
return None
with open(path, 'r', encoding='utf-8') as f:
data = json.load(f)
if not isinstance(data, dict):
return None
required_keys = ['owner', 'repo', 'cached_at', 'data']
if not all(key in data for key in required_keys):
return None
cached_at = datetime.fromisoformat(data['cached_at'])
if datetime.now() - cached_at > self.cache_ttl:
return None
return data
except (json.JSONDecodeError, KeyError, ValueError) as e:
logger.error(f"Cache get error: {e}")
return None
except Exception as e:
logger.error(f"Unexpected cache get error: {e}")
return None
def set(self, owner: str, repo: str, data: Dict):
if not validate_repo_name(owner) or not validate_repo_name(repo):
raise ValueError(f"Invalid repo name: {owner}/{repo}")
temp_path = None
try:
data_size = len(json.dumps(data).encode('utf-8'))
if data_size > self.max_cache_size:
raise ValueError(f"Data too large: {data_size} bytes")
required_keys = ['owner', 'repo', 'analyzed_at']
for key in required_keys:
if key not in data:
raise ValueError(f"Missing required key: {key}")
key = self._get_cache_key(owner, repo)
path = self._get_cache_path(key)
cache_data = {
'owner': owner,
'repo': repo,
'cached_at': datetime.now().isoformat(),
'data': data
}
temp_path = path + '.tmp'
with open(temp_path, 'w', encoding='utf-8') as f:
json.dump(cache_data, f, ensure_ascii=False, indent=2)
f.flush()
os.fsync(f.fileno())
os.rename(temp_path, path)
except Exception as e:
if temp_path and os.path.exists(temp_path):
try:
os.remove(temp_path)
except Exception:
pass
raise
# ============== GitHub Reader v3.2（纯 API 版） ==============
class SecureGitHubReaderV3:
"""GitHub Reader Skill v3.2 — 纯 GitHub REST API，无第三方依赖"""
# 收紧后的触发模式：只接受显式命令
COMMAND_TRIGGERS = [
r'^/github-read\s+([a-zA-Z0-9_-]+/[a-zA-Z0-9._-]+)',
r'github\.com/([a-zA-Z0-9_-]+)/([a-zA-Z0-9._-]+)',
r'^(?:分析|解读|analyze)\s+(?:这个)?(?:仓库|项目)?[:：\s]*(?:https?://github\.com/)?([a-zA-Z0-9_-]+/[a-zA-Z0-9._-]+)',
]
def __init__(self):
self.cache = SecureGitHubReaderCache()
self.semaphore = asyncio.Semaphore(SecurityConfig.MAX_CONCURRENT_REQUESTS)
self._api_call_times: list[float] = []
# ---- 速率限制 ----
async def _rate_limit(self):
"""滑动窗口速率限制 — 确保 API 调用间隔"""
now = time.time()
window = 60.0 # 60 秒窗口
self._api_call_times = [t for t in self._api_call_times if now - t < window]
max_calls_per_window = int(window / SecurityConfig.GITHUB_API_DELAY)
if len(self._api_call_times) >= max_calls_per_window:
wait = self._api_call_times[0] + window - now + 0.1
if wait > 0:
await asyncio.sleep(wait)
self._api_call_times.append(time.time())
# ---- 输入解析 ----
def parse_github_url(self, message: str) -> Optional[tuple[str, str]]:
"""解析 GitHub 仓库标识 — 多模式匹配"""
for pattern in self.COMMAND_TRIGGERS:
match = re.search(pattern, message, re.IGNORECASE)
if match:
groups = match.groups()
if len(groups) >= 2:
owner, repo = groups[0], groups[1]
else:
# 单捕获组（owner/repo 在一起）
parts = groups[0].split('/')
if len(parts) == 2:
owner, repo = parts
else:
continue
repo = repo.rstrip('.git') # 去掉 .git 后缀
if validate_repo_name(owner) and validate_repo_name(repo):
return owner, repo
return None
# ---- API 调用 ----
async def _api_get(self, url: str) -> Optional[Dict]:
"""通用 API GET — 带速率限制、超时、大小限制"""
await self._rate_limit()
import httpx
try:
async with httpx.AsyncClient(timeout=SecurityConfig.GITHUB_API_TIMEOUT) as client:
resp = await client.get(
url,
headers={
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
"User-Agent": "github-reader-skill-v3.2",
},
)
if resp.status_code != 200:
logger.warning(f"GitHub API {url} returned {resp.status_code}")
return None
data = resp.json()
if not isinstance(data, dict):
return None
return data
except Exception as e:
logger.error(f"GitHub API call failed: {e}")
return None
async def fetch_repo_info(self, owner: str, repo: str) -> Optional[Dict]:
"""获取仓库基础信息"""
if not validate_repo_name(owner) or not validate_repo_name(repo):
return None
async with self.semaphore:
data = await self._api_get(
safe_url_join('https://api.github.com/repos', owner, repo)
)
if not data:
return None
license_info = data.get('license')
return {
'stars': self.format_number(data.get('stargazers_count', 0)),
'forks': self.format_number(data.get('forks_count', 0)),
'issues': data.get('open_issues_count', 0),
'language': data.get('language', 'Unknown'),
'license': license_info.get('spdx_id', 'Unknown') if license_info else 'Unknown',
'description': (data.get('description') or '暂无描述')[:500],
'topics': (data.get('topics') or [])[:20],
'updated': self.relative_time(data.get('pushed_at', '')),
'homepage': (data.get('homepage') or '')[:200],
'size_kb': data.get('size', 0),
'archived': data.get('archived', False),
'default_branch': data.get('default_branch', 'main'),
'created': data.get('created_at', ''),
}
async def fetch_readme(self, owner: str, repo: str) -> Optional[str]:
"""获取 README 内容（截取前 3000 字符用于摘要）"""
if not validate_repo_name(owner) or not validate_repo_name(repo):
return None
async with self.semaphore:
data = await self._api_get(
safe_url_join('https://api.github.com/repos', owner, repo, 'readme')
)
if not data:
return None
content = data.get('content', '')
if not content:
return None
try:
import base64
decoded = base64.b64decode(content).decode('utf-8', errors='replace')
return decoded[:3000]
except Exception:
return None
# ---- 分析入口 ----
async def analyze_project(self, owner: str, repo: str) -> Dict:
"""综合分析项目 — 纯 GitHub API"""
if not validate_repo_name(owner) or not validate_repo_name(repo):
raise ValueError(f"Invalid repo name: {owner}/{repo}")
# 1. 检查缓存
cached = self.cache.get(owner, repo)
if cached:
cached_data = cached.get('data', cached)
cached_data['from_cache'] = True
return cached_data
# 2. 并行拉取
repo_task = asyncio.create_task(self.fetch_repo_info(owner, repo))
readme_task = asyncio.create_task(self.fetch_readme(owner, repo))
repo_info, readme_raw = await asyncio.gather(
repo_task, readme_task, return_exceptions=True
)
if isinstance(repo_info, Exception):
logger.error(f"Repo info fetch failed: {repo_info}")
repo_info = {}
if isinstance(readme_raw, Exception):
logger.error(f"README fetch failed: {readme_raw}")
readme_raw = None
# 3. 生成报告
report = self.build_report(owner, repo, repo_info, readme_raw)
# 4. 缓存
if report.get('success'):
try:
self.cache.set(owner, repo, report)
except Exception as e:
logger.error(f"Failed to cache result: {e}")
report['from_cache'] = False
return report
def build_report(
self,
owner: str,
repo: str,
repo_info: Dict,
readme_raw: Optional[str],
) -> Dict:
"""构建分析报告 — 纯本地生成，不经过任何第三方"""
info = repo_info or {}
github_url = safe_url_join('https://github.com', owner, repo)
analyzed_at = datetime.now().isoformat()
report = {
'owner': owner,
'repo': repo,
'analyzed_at': analyzed_at,
'github_url': github_url,
'github_info': info,
'success': True,
}
# 从 README 提取摘要
summary_lines = []
if readme_raw:
for line in readme_raw.split('\n'):
clean = line.strip()
if clean and not clean.startswith('#') and not clean.startswith('![') \
and not clean.startswith('<') and len(clean) > 20:
summary_lines.append(clean[:200])
if len(summary_lines) >= 5:
break
report['readme_snippets'] = summary_lines
# 隐私声明
report['privacy_notice'] = SecurityConfig.PRIVACY_NOTICE.format(
api_url=safe_url_join('https://api.github.com/repos', owner, repo),
cache_ttl=SecurityConfig.CACHE_TTL_HOURS,
cache_dir=SecurityConfig.CACHE_DIR,
)
# Markdown 报告
report['markdown'] = self._render_markdown(report)
return report
def _render_markdown(self, report: Dict) -> str:
owner = report['owner']
repo = report['repo']
info = report.get('github_info', {})
snippets = report.get('readme_snippets', [])
github_url = report['github_url']
description = info.get('description', '这是一个开源项目。')
snippet_block = '\n'.join(f'> {s}' for s in snippets) if snippets else '> *（README 暂无摘要）*'
archived_note = '\n> ⚠️ **注意**：此仓库已被归档（只读）\n' if info.get('archived') else ''
return f"""# 📦 {owner}/{repo} 深度解读报告
> **分析时间**: {datetime.now().strftime('%Y-%m-%d %H:%M')}
> **数据来源**: GitHub REST API（纯 API，不经第三方）
> **隐私**: 仅使用 GitHub 官方 API，不将数据发送给任何第三方服务
{archived_note}
---
## 💡 项目简介
{description}
## 📊 项目卡片
| 指标 | 值 |
|------|-----|
| ⭐ Stars | {info.get('stars', 'N/A')} |
| 🍴 Forks | {info.get('forks', 'N/A')} |
| 📝 Issues | {info.get('issues', 'N/A')} |
| 🐍 语言 | {info.get('language', 'Unknown')} |
| 📄 许可证 | {info.get('license', 'Unknown')} |
| 🕐 最后更新 | {info.get('updated', 'N/A')} |
| 🗂️ 仓库大小 | {info.get('size_kb', 'N/A')} KB |
| 🌿 默认分支 | `{info.get('default_branch', 'main')}` |
| 🏷️ 主题标签 | {' '.join(f'`{t}`' for t in info.get('topics', [])) if info.get('topics') else '无'} |
## 🔗 链接
| 平台 | 链接 |
|------|------|
| **GitHub** | {github_url} |
| 主页 | {info.get('homepage') or '无'} |
## 📖 README 摘要
{snippet_block}
## 🔗 快速开始
```bash
git clone {github_url}.git
cd {repo}
```
{report.get('privacy_notice', '')}
---
*v3.2 — 纯 GitHub API，无第三方数据外传*
"""
# ---- 工具函数 ----
def format_number(self, num: int) -> str:
if num >= 1000000:
return f'{num / 1000000:.1f}M'
elif num >= 1000:
return f'{num / 1000:.1f}k'
return str(num)
def relative_time(self, date_str: str) -> str:
if not date_str:
return 'N/A'
try:
from datetime import timezone
date = datetime.fromisoformat(date_str.replace('Z', '+00:00'))
diff = (datetime.now(timezone.utc) - date).days
if diff == 0:
return '今天'
elif diff == 1:
return '昨天'
elif diff < 7:
return f'{diff}天前'
elif diff < 30:
return f'{diff // 7}周前'
elif diff < 365:
return f'{diff // 30}个月前'
else:
return f'{diff // 365}年前'
except Exception:
return 'N/A'
# ============== Skill 入口 ==============
async def run(context):
"""Skill 入口函数 — v3.2 纯 GitHub API 版"""
try:
message = context.get('message', '')
reader = SecureGitHubReaderV3()
target = reader.parse_github_url(message)
if not target:
return {
'error': '未能识别 GitHub 仓库链接',
'hint': (
'支持格式：\n'
'• `/github-read owner/repo`\n'
'• `https://github.com/owner/repo`\n'
'• `分析 owner/repo`'
),
'success': False,
}
owner, repo = target
result = await reader.analyze_project(owner, repo)
return {
'report': result.get('markdown', ''),
'from_cache': result.get('from_cache', False),
'success': result.get('success', False),
}
except Exception as e:
logger.error(f"Skill execution failed: {e}")
return {
'error': f'分析失败：{str(e)[:200]}',
'success': False,
}