039_密码学实战：零知识证明技术深度解析——从交互式协议到zkSNARK实现的完整指南

协议执行过程：
1. 证明者知道图同构映射f: V(G1) → V(G2)
2. 证明者随机选择一个图H，它与G1和G2同构，通过随机置换π生成
3. 证明者将H发送给验证者
4. 验证者随机选择b ∈ {1, 2}
5. 如果b=1，证明者发送π给验证者，验证H与G1同构
   如果b=2，证明者发送π∘f⁻¹给验证者，验证H与G2同构
6. 验证者检查证明者的响应是否正确
7. 重复上述步骤多次以降低欺骗概率

import numpy as np
import random
from itertools import permutations

def generate_random_graph(n, p=0.5):
    """生成随机图的邻接矩阵"""
    graph = np.zeros((n, n), dtype=int)
    for i in range(n):
        for j in range(i+1, n):
            if random.random() < p:
                graph[i][j] = 1
                graph[j][i] = 1
    return graph

def apply_permutation(graph, perm):
    """应用置换到图上"""
    n = len(graph)
    new_graph = np.zeros((n, n), dtype=int)
    for i in range(n):
        for j in range(n):
            new_graph[perm[i]][perm[j]] = graph[i][j]
    return new_graph

def verify_isomorphism(graph1, graph2, perm):
    """验证置换是否是两个图之间的同构映射"""
    n = len(graph1)
    if len(graph2) != n or len(perm) != n:
        return False
    
    # 检查perm是否是一个置换
    if sorted(perm) != list(range(n)):
        return False
    
    # 应用置换并验证
    transformed = apply_permutation(graph1, perm)
    return np.array_equal(transformed, graph2)

class Prover:
    def __init__(self, graph1, graph2, isomorphism):
        self.graph1 = graph1
        self.graph2 = graph2
        self.isomorphism = isomorphism  # G1 -> G2的同构映射
        self.n = len(graph1)
    
    def create_challenge(self):
        # 随机生成一个置换
        self.random_perm = random.sample(range(self.n), self.n)
        # 应用随机置换生成H
        self.H = apply_permutation(self.graph1, self.random_perm)
        return self.H
    
    def respond_to_verifier(self, challenge):
        if challenge == 1:
            # 证明H与G1同构
            return self.random_perm
        elif challenge == 2:
            # 证明H与G2同构 (使用同构映射的逆)
            # 计算同构映射的逆
            inverse_isomorphism = [0] * self.n
            for i in range(self.n):
                inverse_isomorphism[self.isomorphism[i]] = i
            
            # 组合随机置换和逆同构映射
            combined = [inverse_isomorphism[self.random_perm[i]] for i in range(self.n)]
            return combined
        else:
            raise ValueError("挑战必须是1或2")

class Verifier:
    def __init__(self, graph1, graph2):
        self.graph1 = graph1
        self.graph2 = graph2
    
    def generate_challenge(self):
        # 随机选择1或2
        self.challenge = random.choice([1, 2])
        return self.challenge
    
    def verify_response(self, H, response):
        if self.challenge == 1:
            # 验证H与G1同构
            return verify_isomorphism(self.graph1, H, response)
        elif self.challenge == 2:
            # 验证H与G2同构
            return verify_isomorphism(self.graph2, H, response)
        else:
            return False

def run_isomorphism_zkp_protocol(n=5, rounds=20):
    """运行图同构零知识证明协议"""
    # 生成随机图G1
    G1 = generate_random_graph(n)
    
    # 生成同构映射并创建G2
    isomorphism = random.sample(range(n), n)
    G2 = apply_permutation(G1, isomorphism)
    
    # 初始化证明者和验证者
    prover = Prover(G1, G2, isomorphism)
    verifier = Verifier(G1, G2)
    
    # 运行多轮协议
    accepted_rounds = 0
    for i in range(rounds):
        # 证明者创建挑战
        H = prover.create_challenge()
        
        # 验证者生成挑战
        challenge = verifier.generate_challenge()
        
        # 证明者响应
        response = prover.respond_to_verifier(challenge)
        
        # 验证者验证
        if verifier.verify_response(H, response):
            accepted_rounds += 1
    
    # 计算接受概率
    acceptance_rate = accepted_rounds / rounds
    print(f"协议执行{rounds}轮，接受率：{acceptance_rate:.2f}")
    
    # 如果接受率高，验证者可以确信两个图同构
    return acceptance_rate > 0.9

# 运行协议示例
if __name__ == "__main__":
    result = run_isomorphism_zkp_protocol(n=5, rounds=20)
    print(f"验证者结论：两个图{'是' if result else '不是'}同构的")

import hashlib

def fiat_shamir_transform(interactive_protocol, public_inputs):
    """执行Fiat-Shamir变换，将交互式协议转换为非交互式"""
    # 初始化
    prover = interactive_protocol['prover']
    verifier_check = interactive_protocol['verifier_check']
    
    # 1. 证明者生成初始承诺
    commitment = prover.generate_commitment()
    
    # 2. 计算哈希作为挑战
    challenge_input = commitment + str(public_inputs).encode()
    challenge = hashlib.sha256(challenge_input).hexdigest()[:8]  # 使用哈希值的前8个字符
    
    # 3. 生成响应
    response = prover.generate_response(challenge)
    
    # 4. 返回非交互式证明
    proof = (commitment, response)
    
    # 5. 验证（可选）
    is_valid = verifier_check(public_inputs, commitment, challenge, response)
    
    return proof, is_valid

import hashlib
import random
import math

def is_prime(n):
    """检查一个数是否为素数"""
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

def generate_safe_prime(bit_length=256):
    """生成安全素数（p-1有大素数因子）"""
    while True:
        q = random.getrandbits(bit_length - 1)
        if is_prime(q):
            p = 2 * q + 1
            if is_prime(p):
                return p

def generate_group_parameters(bit_length=256):
    """生成循环群参数"""
    p = generate_safe_prime(bit_length)
    # 寻找原根
    while True:
        g = random.randint(2, p-1)
        if pow(g, 2, p) != 1 and pow(g, (p-1)//2, p) != 1:
            return p, g

class NIZKProof:
    def __init__(self, p, g):
        self.p = p  # 素数模数
        self.g = g  # 生成元
    
    def prove(self, x, y):
        """证明者证明知道x，使得g^x ≡ y mod p"""
        # 检查陈述是否为真
        if pow(self.g, x, self.p) != y:
            raise ValueError("陈述为假，无法生成证明")
        
        # 1. 选择随机数r
        r = random.randint(1, self.p - 2)
        
        # 2. 计算承诺t = g^r mod p
        t = pow(self.g, r, self.p)
        
        # 3. 使用Fiat-Shamir变换生成挑战c
        # 将公共参数和承诺组合起来作为哈希输入
        challenge_data = f"{self.p}:{self.g}:{y}:{t}".encode()
        c = int(hashlib.sha256(challenge_data).hexdigest(), 16) % (self.p - 1)
        
        # 4. 计算响应s = r + c*x mod (p-1)
        s = (r + c * x) % (self.p - 1)
        
        # 返回非交互式证明
        return (t, s)
    
    def verify(self, y, proof):
        """验证者验证证明"""
        t, s = proof
        
        # 1. 重新计算挑战c
        challenge_data = f"{self.p}:{self.g}:{y}:{t}".encode()
        c = int(hashlib.sha256(challenge_data).hexdigest(), 16) % (self.p - 1)
        
        # 2. 验证g^s ≡ t * y^c mod p
        lhs = pow(self.g, s, self.p)
        rhs = (t * pow(y, c, self.p)) % self.p
        
        return lhs == rhs

# 示例使用
def run_nizk_example():
    # 生成参数
    print("生成循环群参数...")
    p, g = generate_group_parameters(128)  # 为了演示使用较小的参数
    print(f"p = {p}")
    print(f"g = {g}")
    
    # 初始化NIZK系统
    nizk = NIZKProof(p, g)
    
    # 证明者选择秘密x
    x = random.randint(1, p-2)
    print(f"证明者的秘密x = {x}")
    
    # 计算公开值y = g^x mod p
    y = pow(g, x, p)
    print(f"公开值y = {y}")
    
    # 生成非交互式证明
    print("生成非交互式零知识证明...")
    proof = nizk.prove(x, y)
    print(f"证明: t = {proof[0]}, s = {proof[1]}")
    
    # 验证证明
    print("验证证明...")
    is_valid = nizk.verify(y, proof)
    print(f"证明{'有效' if is_valid else '无效'}")
    
    # 测试伪造证明
    print("测试伪造证明...")
    fake_proof = (random.randint(1, p-1), random.randint(1, p-2))
    is_fake_valid = nizk.verify(y, fake_proof)
    print(f"伪造证明{'被错误接受' if is_fake_valid else '被正确拒绝'}")

if __name__ == "__main__":
    run_nizk_example()

import numpy as np
from collections import defaultdict

class SimpleQAP:
    def __init__(self):
        self.wires = {}
        self.current_wire_id = 1
        self.constraints = []
    
    def add_wire(self, is_input=False, is_output=False):
        """添加新的线路"""
        wire_id = self.current_wire_id
        self.wires[wire_id] = {'is_input': is_input, 'is_output': is_output}
        self.current_wire_id += 1
        return wire_id
    
    def add_addition_constraint(self, a, b, c):
        """添加加法约束：a + b = c"""
        self.constraints.append(('add', a, b, c))
    
    def add_multiplication_constraint(self, a, b, c):
        """添加乘法约束：a * b = c"""
        self.constraints.append(('mul', a, b, c))
    
    def generate_qap(self):
        """生成二次算术程序"""
        n = len(self.wires)
        m = len(self.constraints)
        
        # 初始化多项式系数矩阵
        A = np.zeros((n+1, m+1))  # A矩阵
        B = np.zeros((n+1, m+1))  # B矩阵
        C = np.zeros((n+1, m+1))  # C矩阵
        
        # 添加约束到QAP
        for i, (constraint_type, a, b, c) in enumerate(self.constraints):
            if constraint_type == 'add':
                A[a, i+1] = 1
                A[b, i+1] = 1
                C[c, i+1] = 1
            elif constraint_type == 'mul':
                A[a, i+1] = 1
                B[b, i+1] = 1
                C[c, i+1] = 1
        
        # 添加常数项（对应线路0）
        A[0, 0] = 1
        B[0, 0] = 1
        C[0, 0] = 1
        
        return A, B, C

class SimpleZKSNARK:
    def __init__(self, qap_system):
        self.qap = qap_system
        self.A, self.B, self.C = qap_system.generate_qap()
        self.n = self.A.shape[0] - 1  # 线路数量
        self.m = self.A.shape[1] - 1  # 约束数量
        
        # 生成随机参数（在实际系统中，这需要可信设置）
        self.tau = np.random.randint(1, 1000)
        self.alpha = np.random.randint(1, 1000)
        self.beta = np.random.randint(1, 1000)
        self.gamma = np.random.randint(1, 1000)
        self.delta = np.random.randint(1, 1000)
    
    def evaluate_polynomial(self, coeffs, x):
        """在点x处评估多项式"""
        result = 0
        for i, coeff in enumerate(coeffs):
            result += coeff * (x ** i)
        return result
    
    def trusted_setup(self):
        """执行可信设置过程（简化版）"""
        # 计算公共参数
        pp = {}
        
        # A多项式评估
        pp['A'] = []
        for i in range(self.n + 1):
            eval_A = self.evaluate_polynomial(self.A[i], self.tau)
            pp['A'].append(eval_A)
        
        # B多项式评估
        pp['B'] = []
        for i in range(self.n + 1):
            eval_B = self.evaluate_polynomial(self.B[i], self.tau)
            pp['B'].append(eval_B)
        
        # C多项式评估
        pp['C'] = []
        for i in range(self.n + 1):
            eval_C = self.evaluate_polynomial(self.C[i], self.tau)
            pp['C'].append(eval_C)
        
        # 添加其他参数
        pp['alpha'] = self.alpha
        pp['beta'] = self.beta
        pp['gamma'] = self.gamma
        pp['delta'] = self.delta
        pp['tau'] = self.tau
        
        # 生成证明密钥和验证密钥
        proving_key = {
            'A': pp['A'], 'B': pp['B'], 'C': pp['C'],
            'beta': pp['beta'], 'gamma': pp['gamma'],
            'delta': pp['delta'], 'tau': pp['tau']
        }
        
        verification_key = {
            'A_query': [pp['alpha'] * pp['A'][i] for i in range(self.n + 1)],
            'B_query': [pp['beta'] * pp['B'][i] for i in range(self.n + 1)],
            'C_query': pp['C'],
            'gamma': pp['gamma'],
            'delta': pp['delta']
        }
        
        return proving_key, verification_key
    
    def generate_witness(self, input_values):
        """生成witness（所有线路的赋值）"""
        # 初始化witness数组
        witness = np.zeros(self.n + 1)
        witness[0] = 1  # 常数项
        
        # 设置输入值
        input_count = 0
        for i in range(1, self.n + 1):
            if i in self.qap.wires and self.qap.wires[i]['is_input']:
                if input_count < len(input_values):
                    witness[i] = input_values[input_count]
                    input_count += 1
        
        # 计算中间值（简化版，实际需要拓扑排序）
        # 这里我们假设约束已经按正确顺序排列
        for constraint_type, a, b, c in self.qap.constraints:
            if constraint_type == 'add':
                witness[c] = witness[a] + witness[b]
            elif constraint_type == 'mul':
                witness[c] = witness[a] * witness[b]
        
        return witness
    
    def prove(self, proving_key, witness):
        """生成零知识证明"""
        # 选择随机数
        r = np.random.randint(1, 1000)
        s = np.random.randint(1, 1000)
        
        # 计算A份额
        A_share = proving_key['alpha']
        for i in range(1, self.n + 1):
            A_share += witness[i] * proving_key['A'][i]
        A_share += r * proving_key['delta']
        
        # 计算B份额
        B_share = proving_key['beta']
        for i in range(1, self.n + 1):
            B_share += witness[i] * proving_key['B'][i]
        B_share += s * proving_key['delta']
        
        # 计算C份额（简化版）
        C_share = 0
        for i in range(1, self.n + 1):
            C_share += witness[i] * proving_key['C'][i]
        C_share += (r * proving_key['beta'] + s * proving_key['alpha'] + r * s * proving_key['delta']) / proving_key['gamma']
        
        return (A_share, B_share, C_share)
    
    def verify(self, verification_key, input_values, proof):
        """验证零知识证明"""
        A_share, B_share, C_share = proof
        
        # 计算验证方程的左侧
        left = A_share * B_share
        
        # 计算验证方程的右侧
        right = verification_key['C_query'][0]
        
        # 添加输入部分
        input_count = 0
        for i in range(1, self.n + 1):
            if i in self.qap.wires and self.qap.wires[i]['is_input']:
                if input_count < len(input_values):
                    right += input_values[input_count] * verification_key['C_query'][i]
                    input_count += 1
        
        # 添加其他项
        right += (verification_key['gamma'] * C_share)
        
        # 验证等式是否成立
        return np.isclose(left, right)

# 示例：证明知道x，使得x^3 + x + 5 = 35
def run_simple_zk_snark_example():
    # 创建QAP系统
    qap = SimpleQAP()
    
    # 添加输入线路
    x = qap.add_wire(is_input=True)
    
    # 添加中间线路
    x_squared = qap.add_wire()
    x_cubed = qap.add_wire()
    x_cubed_plus_x = qap.add_wire()
    
    # 添加输出线路
    output = qap.add_wire(is_output=True)
    
    # 添加常数5
    const_5 = qap.add_wire()
    
    # 添加约束
    qap.add_multiplication_constraint(x, x, x_squared)  # x^2
    qap.add_multiplication_constraint(x, x_squared, x_cubed)  # x^3
    qap.add_addition_constraint(x_cubed, x, x_cubed_plus_x)  # x^3 + x
    qap.add_addition_constraint(x_cubed_plus_x, const_5, output)  # x^3 + x + 5
    
    # 创建zkSNARK系统
    zksnark = SimpleZKSNARK(qap)
    
    # 执行可信设置
    proving_key, verification_key = zksnark.trusted_setup()
    
    # 生成witness（x=3是解，因为3^3 + 3 + 5 = 27 + 3 + 5 = 35）
    witness = zksnark.generate_witness([3, 5])  # x=3, const_5=5
    
    # 生成证明
    proof = zksnark.prove(proving_key, witness)
    print(f"生成的证明: {proof}")
    
    # 验证证明
    is_valid = zksnark.verify(verification_key, [3, 5], proof)
    print(f"证明验证结果: {'有效' if is_valid else '无效'}")

if __name__ == "__main__":
    run_simple_zk_snark_example()

import hashlib
import random
import math

def solve_zkp_challenge():
    """解决CTF中常见的离散对数零知识证明挑战"""
    # 假设以下参数从挑战中获取
    p = 0x8542D69E4C044F18E8B92435BF6FF7DE457283915C45517D722EDB8B08F1DFC3  # 大素数
    g = 2  # 生成元
    y = 0x7D7AE474569F411E64D8B4580C84A3E5C483626A6C4D57FB8C8F24D01C48A148  # 公钥 y = g^x mod p
    
    # 挑战描述：服务器实现了一个有缺陷的零知识证明协议，尝试恢复私钥x
    
    # 假设服务器的证明验证函数有以下漏洞：
    # 它不验证响应s是否满足 g^s ≡ t * y^c mod p，而只检查部分条件
    
    # 漏洞利用：我们可以通过选择特定的t和c值来获取有关x的信息
    
    # 方法1：重放攻击
    # 如果服务器接受相同的证明多次，我们可以重放有效的证明
    
    # 方法2：特殊参数攻击
    # 选择t = g^r，c = 0，那么s = r，验证会通过但不会泄露信息
    
    # 方法3：针对实现漏洞的攻击
    # 假设服务器在生成c时使用了可预测的随机性
    
    # 模拟服务器的行为
    def verify_proof(t, c, s):
        # 有缺陷的验证函数，只检查部分条件
        return pow(g, s, p) == (t * pow(y, c, p)) % p
    
    # 尝试利用漏洞恢复x
    # 例如：如果我们能让服务器使用我们选择的c值
    
    # 假设我们可以选择c=1，然后获取正确的s值
    # 这将允许我们计算 t = g^s / y mod p
    # 如果我们还能让服务器选择特定的r，我们可以建立方程求解x
    
    # 这里我们模拟一个简化的攻击，假设我们能够观察多组(t, c, s)值
    observations = []
    
    # 假设我们观察到以下证明
    for i in range(10):
        # 模拟观察到的有效证明
        r = random.randint(1, p-2)
        t = pow(g, r, p)
        c = random.randint(1, p-2)  # 这里假设c是随机的
        s = (r + c * 42) % (p-1)  # 假设x=42
        observations.append((t, c, s))
    
    # 从观察中恢复x
    # 对于两个观察 (t1, c1, s1) 和 (t2, c2, s2)
    # s1 = r1 + c1*x mod (p-1)
    # s2 = r2 + c2*x mod (p-1)
    # 如果我们知道 r1 = log_g(t1), r2 = log_g(t2)
    # 但计算离散对数本身就是困难的，所以这需要其他漏洞
    
    # 在实际的CTF挑战中，可能会有侧信道信息泄露，或者协议实现有缺陷
    # 例如：如果c使用了确定性生成，或者服务器泄露了部分中间状态
    
    print("CTF零知识证明挑战分析完成")
    print("在实际挑战中，需要根据具体的协议实现和漏洞进行针对性攻击")
    print("常见的漏洞包括：可预测的随机性、错误的参数验证、侧信道信息泄露等")

# 运行示例
if __name__ == "__main__":
    solve_zkp_challenge()