社会工程

DeeLMind大约 4 分钟

社会工程

什么是社会工程

社会工程是一种通过操纵人的心理来获取信息、欺诈或入侵系统的行为。它利用人们的信任、同情心、恐惧感或其他心理弱点,诱使他们做出违反安全策略或泄露敏感信息的举动。

社会工程攻击可以采取多种形式,包括:

  • 网络钓鱼:攻击者会发送伪装成合法机构或个人的电子邮件或短信,诱使受害者点击恶意链接或泄露个人信息。

  • 电话诈骗:攻击者会冒充政府部门、银行或其他机构的工作人员,通过电话联系受害者,并要求他们提供个人信息或转账汇款。

个人信息:姓名、身份证号码、银行账户信息、信用卡号等。 机密信息:商业机密、政府文件、军事秘密等。 计算机系统:获取访问权限、植入恶意软件、窃取数据等。 社会工程攻击之所以有效,是因为它利用了人们的心理弱点。例如,人们通常会信任看起来合法或权威的人,也容易被恐惧感或紧迫感所驱使。

什么是社工库

社工库,也称为社会工程数据库,是指包含大量个人信息的数据库,这些信息通常是通过社会工程攻击、网络攻击或数据泄露等方式收集的。社工库中的信息可能包括:

个人身份信息:姓名、身份证号码、地址、电话号码、邮箱地址等。 账户信息:社交媒体账号、银行账户、信用卡信息等。 其他敏感信息:密码、聊天记录、照片、旅行记录等。

开源社工库

电脑端打开本网站首页,限时查看右边数据查询。

搭建社工库 - 直播讲解open in new window

  1. 数据准备
magnet:?xt=urn:btih:e831fcfaca5f0208009406b7b090014cef9228a9&dn=passwds

magnet:?xt=urn:btih:963fd90eee4db809ed4224d1ca7a0639c443cf4b

magnet:?xt=urn:btih:c81e0644fd67f73d81b94a31e3fc726679638a98&dn=pcht-v1
  1. 大数据查询环境(或者用GitHub)

start .\elasticsearch-8.13.0\bin\elasticsearch.bat

elasticsearch-8.13.0\config\elasticsearch.yml

http.port: 9200
http.host: 0.0.0.0
xpack.security.enabled: false
xpack.security.enrollment.enabled: false

start .\kibana-8.13.0\bin\kibana.bat

kibana-8.13.0\config\kibana.yml 为空

.\ngrok http --domain=surely-up-alpaca.ngrok-free.app 5601

  1. 导入数据即可
import pandas as pd
from elasticsearch import Elasticsearch
import os
import hashlib
import json

# Elasticsearch索引名称前缀
index_prefix = 'socialdb'

# CSV文件路径
directory = r"your_path"

# Elasticsearch连接
es = Elasticsearch(['http://127.0.0.1:9200'])

# 记录最后处理的文件及行数的状态文件路径
status_file = "status.json"

# 如果状态文件存在,则加载上次的处理状态
if os.path.exists(status_file):
    with open(status_file, 'r') as f:
        status = json.load(f)
else:
    status = {"last_processed_file": "", "last_processed_row": 0}

# 遍历目录下的所有文件
for root, dirs, files in os.walk(directory):
    for filename in files:
        file_path = os.path.join(root, filename)
        if file_path < status['last_processed_file']:
            print("跳过", file_path)
            continue  # 跳过已经处理过的文件

        # 创建 SHA-256 哈希对象
        hash_object = hashlib.sha256()

        # 更新哈希对象
        hash_object.update(file_path.encode())

        # 计算哈希值作为索引名称的一部分
        index_name = index_prefix + "_" + hash_object.hexdigest()

        # 记录正在处理的文件
        print(f"Processing file: {file_path}")

        # 读取CSV文件并插入数据到Elasticsearch
        try:
            # 读取CSV文件,将所有列都设定为字符串类型
            df = pd.read_csv(file_path, delimiter='----', encoding='utf-8', header=None, dtype=str)

            # 创建一个列表用于存储批量插入的数据
            bulk_data = []

            for row_index, row in df.iterrows():
                if row_index < status['last_processed_row']:
                    print("跳过", row_index)
                    continue  # 跳过已经处理过的行

                # 要替换的键映射关系
                key_mapping = {0: 'QQ', 1: 'Tel'}

                # 使用字典推导式替换键,并将Tel转换为字符串
                new_dict = {key_mapping.get(old_key, old_key): str(value) for old_key, value in row.to_dict().items()}

                # 将数据加入批量插入列表
                bulk_data.append({"index": {"_index": index_name.lower()}})
                bulk_data.append(new_dict)

                # print(f"正在添加Row {row_index}{new_dict}")

                # 当批量插入列表中的数据量达到一定程度时执行批量插入
                if len(bulk_data) >= 5000:
                    res = es.bulk(index=index_name.lower(), body=bulk_data, refresh=True)
                    print(f"已插入 {len(bulk_data) // 2} 条数据到 Elasticsearch")
                    bulk_data = []  # 清空批量插入列表

            # 处理剩余未插入的数据
            if bulk_data:
                res = es.bulk(index=index_name.lower(), body=bulk_data, refresh=True)
                print(f"已插入 {len(bulk_data) // 2} 条数据到 Elasticsearch")

        except Exception as e:
            print(f"Error reading file {file_path}: {str(e)}")
            # 更新状态文件
            status['last_processed_file'] = file_path
            status['last_processed_row'] = row_index
            with open(status_file, 'w') as f:
                json.dump(status, f)
                exit()

获取TG社工库 - 直播讲解open in new window

from telethon import TelegramClient

# 使用你从 Telegram 获取的 API ID 和 API Hash
api_id = 'your_api_id'
api_hash = 'your_api_hash'

# 创建客户端
client = TelegramClient('session_name', api_id, api_hash)

# 异步函数,获取消息内容
async def main():
    # 登录 Telegram
    await client.start()
    
    # 查找特定的群聊
    chat = await client.get_entity('group_chat_name_or_id')

    # 获取最近的消息
    async for message in client.iter_messages(chat, limit=10):
        print(message.sender_id, message.text)

# 运行脚本
with client:
    client.loop.run_until_complete(main())

  • 客户端程序获取

  • WEB页面获取

上次编辑于:
贡献者: DeeLMind,DeeLMind,DeeLMind