ykrishhh/telegram-security

# Telegram 安全       Telegram 安全监控、bot 检测、频道分析及隐私审计工具。分析 Telegram session，检测恶意 bot，并审计您的消息隐私。 **关键词**：`telegram` `security` `privacy` `bot-detection` `monitoring` `python` `telethon` `pyrogram` `session-security` `api-security` ## 目录 - [为什么开发此项目](#why-this-exists) - [功能](#features) - [安装说明](#installation) - [Bot 检测](#bot-detection) - [频道分析](#channel-analysis) - [隐私审计](#privacy-audit) - [Session 安全](#session-security) - [API 安全](#api-security) - [数据收集风险](#data-collection-risks) - [安全使用指南](#secure-usage-guide) - [Python 脚本](#python-scripts) - [免责声明](#disclaimer) - [许可证](#license) ## 为什么开发此项目 Telegram 在很多方面都很棒，但除非您使用 Secret Chats，否则隐私并不是它的强项。我经常遇到同样的问题：频道里的 bot 垃圾信息、以明文形式存放在磁盘上的 session 文件，以及如果不深入研究 API 文档就无法弄清楚的隐私设置。因此，我开发了一些工具来简化这些事情。 ## 功能 - **Bot 检测** — 识别自动化账户、垃圾 bot 和钓鱼尝试 - **频道分析** — 审计频道元数据、发布模式和订阅者增长情况 - **隐私审计** — 检查您的账户曝光度和设置 - **Session 管理** — 检测并撤销未经授权的 session - **API 安全辅助** — 为 API 凭据和 session 文件提供安全存储 - **数据收集风险评估** — 了解 Telegram 收集了哪些元数据 - **转发追踪检测** — 追踪消息转发链 ## 安装说明 ``` git clone https://github.com/ykrishhh/telegram-security.git cd telegram-security pip install -r requirements.txt ``` ### 前置条件 ``` telethon>=1.30.0 pyrogram>=2.0.0 tgcrypto>=1.2.5 python-dotenv>=1.0.0 cryptography>=41.0.0 rich>=13.0.0 ``` ### 环境设置 ``` # 使用你的凭证创建 .env 文件 cat < .env TELEGRAM_API_ID=your_api_id TELEGRAM_API_HASH=your_api_hash TELEGRAM_PHONE=+1234567890 EOF ``` 从 [my.telegram.org](https://my.telegram.org) 获取 `API_ID` 和 `API_HASH`。 ## Bot 检测 ### 分析用户行为模式 这里的评分系统是启发式的——虽然不完美，但它能抓住大多数自动化账户。时间规律性检查是最可靠的信号。真实的人类不会刚好以 300 秒的间隔发布内容。 ``` #!/usr/bin/env python3 """ Telegram bot detection via behavioral analysis Author: ykrishhh """ import asyncio from datetime import datetime, timedelta from collections import Counter from telethon import TelegramClient from telethon.tl.types import User class BotDetector: """Identify likely automated accounts based on activity patterns.""" def __init__(self, client: TelegramClient): self.client = client async def analyze_user(self, user_id: int) -> dict: """Profile a user for bot-like characteristics.""" user = await self.client.get_entity(user_id) messages = await self.client.get_messages(user, limit=200) if not messages: return {"user_id": user_id, "verdict": "insufficient_data"} # Extract timing patterns timestamps = [msg.date for msg in messages if msg.date] intervals = [] for i in range(1, len(timestamps)): diff = (timestamps[i - 1] - timestamps[i]).total_seconds() intervals.append(abs(diff)) # Calculate metrics avg_interval = sum(intervals) / len(intervals) if intervals else 0 std_dev = self._std_dev(intervals) if intervals else 0 hour_dist = Counter(t.hour for t in timestamps) # Scoring heuristics score = 0 reasons = [] # Regularity check: low std dev = suspicious if avg_interval > 0 and std_dev / avg_interval < 0.1 and len(intervals) > 10: score += 30 reasons.append("Unusually regular posting intervals") # 24/7 activity active_hours = len([h for h, c in hour_dist.items() if c > 2]) if active_hours >= 20: score += 25 reasons.append(f"Active across {active_hours}/24 hours") # No profile photo if not user.photo: score += 10 reasons.append("No profile photo") # Account age check if user.premium is False and user.bot is False: if user.id and user.id > 2000000000: score += 10 reasons.append("Relatively new account ID") # Bot flag already set by Telegram if user.bot: score += 50 reasons.append("Marked as bot by Telegram") verdict = "human" if score >= 60: verdict = "likely_bot" elif score >= 30: verdict = "suspicious" return { "user_id": user_id, "username": getattr(user, "username", None), "first_name": getattr(user, "first_name", ""), "score": score, "verdict": verdict, "reasons": reasons, "metrics": { "avg_post_interval_sec": round(avg_interval, 1), "interval_std_dev": round(std_dev, 1), "active_hours": active_hours, "total_messages": len(messages) } } @staticmethod def _std_dev(values: list) -> float: if len(values) < 2: return 0.0 mean = sum(values) / len(values) variance = sum((x - mean) ** 2 for x in values) / (len(values) - 1) return variance ** 0.5 async def main(): """Scan a list of user IDs for bot behavior.""" api_id = int(os.environ["TELEGRAM_API_ID"]) api_hash = os.environ["TELEGRAM_API_HASH"] async with TelegramClient("scanner", api_id, api_hash) as client: detector = BotDetector(client) test_users = [123456789, 987654321] # Replace with actual IDs for uid in test_users: result = await detector.analyze_user(uid) print(f"\nUser {uid}: {result['verdict'].upper()}") print(f" Score: {result['score']}/100") for reason in result.get("reasons", []): print(f" - {reason}") if __name__ == "__main__": asyncio.run(main()) ``` ## 频道分析 ### 元数据与增长审计器 此工具会提取频道元数据和最近的消息，以寻找虚假订阅者或自动化发布的迹象。如果一个频道有 10 万关注者，但每个帖子只有 50 次浏览，那就不太对劲了。 ``` #!/usr/bin/env python3 """ Telegram channel security and metadata analysis Author: ykrishhh """ import asyncio import os from datetime import datetime from telethon import TelegramClient from telethon.tl.types import ChannelParticipantsRecentlyOffline class ChannelAuditor: """Analyze a Telegram channel for security and integrity.""" def __init__(self, client: TelegramClient): self.client = client async def audit(self, channel_username: str) -> dict: """Run a full audit on the target channel.""" channel = await self.client.get_entity(channel_username) report = { "name": channel.title, "username": channel_username, "id": channel.id, "type": "supergroup" if channel.megagroup else "channel", "subscribers": channel.participants_count, "created": str(channel.date), "verified": getattr(channel, "verified", False), "scam": getattr(channel, "scam", False), "fake": getattr(channel, "fake", False), "signatures": getattr(channel, "signatures", False), "restricted": getattr(channel, "restricted", False), } # Fetch recent messages for pattern analysis messages = [] async for msg in self.client.iter_messages(channel, limit=100): messages.append(msg) if messages: report["message_count"] = len(messages) report["date_range"] = { "oldest": str(messages[-1].date), "newest": str(messages[0].date) } report["forwarded_count"] = sum(1 for m in messages if m.forward) report["media_count"] = sum(1 for m in messages if m.media) report["avg_views"] = ( sum(m.views for m in messages if m.views) // len(messages) if any(m.views for m in messages) else 0 ) # Detect sudden subscriber spikes (buying subscribers) if channel.participants_count and channel.participants_count > 10000: report["subscriber_risk"] = "manual_review_recommended" else: report["subscriber_risk"] = "low" return report async def run(): api_id = int(os.environ["TELEGRAM_API_ID"]) api_hash = os.environ["TELEGRAM_API_HASH"] async with TelegramClient("auditor", api_id, api_hash) as client: auditor = ChannelAuditor(client) result = await auditor.audit("durov") # Example channel print(f"\nChannel: {result['name']}") print(f"Type: {result['type']}") print(f"Subscribers: {result.get('subscribers', 'N/A')}") print(f"Verified: {result['verified']}") print(f"Scam flag: {result['scam']}") print(f"Fake flag: {result['fake']}") print(f"Forwarded messages: {result.get('forwarded_count', 0)}") print(f"Avg views: {result.get('avg_views', 0)}") if __name__ == "__main__": asyncio.run(run()) ``` ## 隐私审计 ### 账户曝光度检查 大多数人没有意识到，他们的手机号码默认对所有人可见。此脚本会检查您的所有隐私设置，并标记出有风险的设置。 ``` #!/usr/bin/env python3 """ Telegram privacy settings audit Author: ykrishhh """ import asyncio import os from telethon import TelegramClient from telethon.tl.functions.account import GetPrivacyRequest from telethon.tl.types import ( PrivacyKeyStatusTimestamp, PrivacyKeyChatInvite, PrivacyKeyPhoneCall, PrivacyKeyPhoneP2P, PrivacyKeyProfilePhoto, PrivacyKeyPhoneNumber, ) class PrivacyAuditor: """Check and report on Telegram privacy settings.""" PRIVACY_KEYS = { PrivacyKeyStatusTimestamp: "Last seen", PrivacyKeyChatInvite: "Groups and channels", PrivacyKeyPhoneCall: "Phone calls", PrivacyKeyPhoneP2P: "Peer-to-peer calls", PrivacyKeyProfilePhoto: "Profile photo", PrivacyKeyPhoneNumber: "Phone number", } def __init__(self, client: TelegramClient): self.client = client async def audit_privacy(self) -> list: """Retrieve and evaluate all privacy settings.""" results = [] for key_cls, label in self.PRIVACY_KEYS.items(): try: result = await self.client(GetPrivacyRequest(key_cls())) privacy_rule = result.rules[0] if result.rules else None if privacy_rule is None: status = "unknown" elif hasattr(privacy_rule, "allow_users"): status = "custom" elif hasattr(privacy_rule, "disallow_users"): status = "restricted" else: status = str(privacy_rule) results.append({ "setting": label, "rule": status, "allows_chat_members": getattr(result, "chats", []) }) except Exception as e: results.append({"setting": label, "rule": f"error: {e}"}) return results def evaluate_risks(self, settings: list) -> list: """Flag privacy settings that may expose the user.""" risks = [] for s in settings: if s["setting"] == "Phone number" and "Everybody" in str(s["rule"]): risks.append("Phone number visible to everyone") if s["setting"] == "Last seen" and "Everybody" in str(s["rule"]): risks.append("Online status visible to everyone") if s["setting"] == "Groups and channels" and "Everybody" in str(s["rule"]): risks.append("Anyone can add you to groups") return risks async def main(): api_id = int(os.environ["TELEGRAM_API_ID"]) api_hash = os.environ["TELEGRAM_API_HASH"] async with TelegramClient("privacy-check", api_id, api_hash) as client: auditor = PrivacyAuditor(client) settings = await auditor.audit_privacy() risks = auditor.evaluate_risks(settings) print("\n=== Privacy Settings Audit ===\n") for s in settings: print(f" {s['setting']:20s} : {s['rule']}") if risks: print("\n=== Privacy Risks Detected ===\n") for r in risks: print(f" [!] {r}") else: print("\n[+] No obvious privacy risks found.") if __name__ == "__main__": asyncio.run(main()) ``` ## Session 安全 ### 检测并撤销未经授权的 Session 我每个月都会运行一次。堆积起来的 session 数量令人大开眼界——旧手机、您忘记的 Web 客户端、那次您从朋友笔记本电脑上登录的经历。干掉那些您不认识的 session。 ``` #!/usr/bin/env python3 """ Telegram session security manager Author: ykrishhh """ import asyncio import os from telethon import TelegramClient from telethon.tl.functions.account import GetAuthorizationsRequest, DeleteAuthorizationRequest class SessionManager: """List, audit, and revoke Telegram sessions.""" def __init__(self, client: TelegramClient): self.client = client async def list_sessions(self) -> list: """Fetch all active sessions.""" result = await self.client(GetAuthorizationsRequest()) sessions = [] for auth in result.authorizations: sessions.append({ "hash": auth.hash, "device": auth.device_model, "platform": auth.platform, "system_version": auth.system_version, "api_id": auth.api_id, "app_name": auth.app_name, "date_created": str(auth.date_created), "date_active": str(auth.date_active), "ip": auth.ip, "country": auth.country, "is_current": getattr(auth, "current", False), }) return sessions async def audit_sessions(self) -> dict: """Identify suspicious sessions.""" sessions = await self.list_sessions() current = [s for s in sessions if s["is_current"]] others = [s for s in sessions if not s["is_current"]] suspicious = [] for s in others: reasons = [] if not s["ip"]: reasons.append("No IP recorded") if "Unknown" in s["device"]: reasons.append("Unrecognized device") suspicious.append({"session": s, "reasons": reasons}) return { "total_sessions": len(sessions), "current_session": current[0] if current else None, "other_sessions": others, "suspicious": [s for s in suspicious if s["reasons"]], "recommendation": "revoke_unknown" if others else "all_clear" } async def revoke(self, session_hash: int) -> bool: """Terminate a specific session.""" try: await self.client(DeleteAuthorizationRequest(hash=session_hash)) return True except Exception: return False async def revoke_all_except_current(self) -> int: """Kill every session except the one in use.""" sessions = await self.list_sessions() revoked = 0 for s in sessions: if not s["is_current"]: if await self.revoke(s["hash"]): revoked += 1 return revoked async def main(): api_id = int(os.environ["TELEGRAM_API_ID"]) api_hash = os.environ["TELEGRAM_API_HASH"] async with TelegramClient("session-check", api_id, api_hash) as client: mgr = SessionManager(client) audit = await mgr.audit_sessions() print(f"\nActive sessions: {audit['total_sessions']}") for s in audit["other_sessions"]: print(f" Device: {s['device']}") print(f" Platform: {s['platform']}") print(f" IP: {s['ip']}") print(f" Last active: {s['date_active']}") print(f" ---") if audit["suspicious"]: print(f"\n[!] {len(audit['suspicious'])} suspicious session(s) detected") confirm = input("Revoke all non-current sessions? (y/N): ") if confirm.lower() == "y": count = await mgr.revoke_all_except_current() print(f"[+] Revoked {count} session(s)") if __name__ == "__main__": asyncio.run(main()) ``` ## API 安全 ### 凭据存储 您的 API 凭据不应该以明文形式存放在共享服务器的 `.env` 文件中。此工具会在主密码的保护下，使用 AES-256-GCM 对它们进行加密。 ``` #!/usr/bin/env python3 """ Secure storage for Telegram API credentials Author: ykrishhh """ import os import json from pathlib import Path from cryptography.hazmat.primitives.ciphers.aead import AESGCM import hashlib import secrets CREDENTIALS_FILE = Path.home() / ".config" / "tg-security" / "credentials.enc" KEY_DERIVATION_ITERATIONS = 600_000 def store_credentials(api_id: str, api_hash: str, phone: str, master_password: str) -> None: """Encrypt and store Telegram credentials on disk.""" CREDENTIALS_FILE.parent.mkdir(parents=True, exist_ok=True) salt = secrets.token_bytes(16) key = hashlib.pbkdf2_hmac("sha256", master_password.encode(), salt, KEY_DERIVATION_ITERATIONS, 32) data = json.dumps({ "api_id": api_id, "api_hash": api_hash, "phone": phone }).encode() nonce = secrets.token_bytes(12) ct = AESGCM(key).encrypt(nonce, data, None) with open(CREDENTIALS_FILE, "wb") as f: f.write(salt + nonce + ct) os.chmod(CREDENTIALS_FILE, 0o600) print(f"[+] Credentials stored at {CREDENTIALS_FILE}") def load_credentials(master_password: str) -> dict: """Decrypt and return stored credentials.""" with open(CREDENTIALS_FILE, "rb") as f: raw = f.read() salt = raw[:16] nonce = raw[16:28] ct = raw[28:] key = hashlib.pbkdf2_hmac("sha256", master_password.encode(), salt, KEY_DERIVATION_ITERATIONS, 32) pt = AESGCM(key).decrypt(nonce, ct, None) return json.loads(pt) ``` ## 数据收集风险 ### Telegram 收集的内容 | 数据类型 | 是否收集 | 保留期限 | |-----------|-----------|-----------| | 手机号码 | 是 | 账户生命周期 | | 通讯录 | 如果同步 | 直到删除 | | IP 地址 | 每个 session | 最长 12 个月 | | 设备信息 | 每个 session | 账户生命周期 | | 使用元数据 | 是 | 无限期 | | 消息内容 | 云端聊天：是 | 直到删除 | | 消息内容 | Secret chats：否 | 端到端加密 | 这是大多数人不想听到的部分。Telegram 将您的云端聊天内容存储在他们的服务器上。它在传输过程中和静止时都是加密的，但 Telegram 持有密钥。Secret chats 则不同——它们是真正的端到端加密的，并且永远不会触及 Telegram 的服务器。 ### 风险缓解步骤 1. **使用 Secret Chats** 进行敏感对话（E2E 加密，仅限设备） 2. 在隐私设置中**禁用通讯录同步** 3. **使用 VPN** 向 Telegram 服务器隐藏您的 IP 4. 为 Telegram 应用**设置密码锁** 5. 使用强密码**启用两步验证** 6. 每月**审查 session** 并撤销未知的 session ## 安全使用指南 ### 两步验证设置 ``` Settings > Privacy and Security > Two-Step Verification > Set Password ``` 选择一个与其他账户无关的密码。即使攻击者拦截了您的短信验证码，这也能防止账户被接管。 ### Secret Chat 激活 ``` Tap a contact's profile > More (⋮) > Start Secret Chat ``` Secret chats 提供： - 端到端加密（无云端存储） - 自毁消息 - 截图通知 - 仅限设备的消息存储 ### 转发保护 在隐私设置中启用“禁止转发”，这样其他人就无法将转发消息归因于您的账户。 ### 密码锁 ``` Settings > Privacy and Security > Passcode Lock > Enable ``` 为应用添加本地的生物识别或 PIN 码保护。 ## Python 脚本 所有脚本都需要来自 [my.telegram.org](https://my.telegram.org) 的 API 凭据。 ### 运行任意脚本 ``` # 设置环境变量 export TELEGRAM_API_ID=12345678 export TELEGRAM_API_HASH=abcdef1234567890abcdef # 运行 bot detection python scripts/bot_detector.py # 运行 channel audit python scripts/channel_auditor.py # 运行 privacy audit python scripts/privacy_auditor.py # 运行 session security check python scripts/session_manager.py ``` ## 免责声明 这些工具旨在用于审计您自己的账户和分析公开可用的频道。请勿在未经他人明确同意的情况下使用它们来监控其他用户。未经授权的监控违反了 Telegram 的服务条款和适用的隐私法律。作者对滥用行为不承担任何责任。 ## 许可证 MIT 许可证。有关详细信息，请参阅 [LICENSE](LICENSE)。

标签：Python, Telegram, Telethon, 后端开发, 无后门, 机器人检测, 逆向工具, 隐私分析