微软为AI模型划红线：不得形成自身目标、不得掩盖行为——AI行为准则与企业数据信任危机深度解析

// goal_drift_detector.go
// 检测AI模型是否形成自主目标的验证器

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"regexp"
	"strings"
	"time"
)

// ActionRecord 表示模型的一次动作记录
type ActionRecord struct {
	Timestamp    time.Time `json:"timestamp"`
	ActionType   string    `json:"action_type"`
	Description  string    `json:"description"`
	TaskID       string    `json:"task_id"`
	GoalAligned  bool      `json:"goal_aligned"`
	SourceAction string    `json:"source_action,omitempty"`
}

// GoalDriftDetector 目标漂移检测器
type GoalDriftDetector struct {
	assignedTasks  map[string]string   // 任务ID -> 原始目标描述
	actionHistory  []ActionRecord      // 动作历史
	driftThreshold float64             // 漂移阈值
}

// DetectionResult 检测结果
type DetectionResult struct {
	HasDrift       bool     `json:"has_drift"`
	DriftScore     float64  `json:"drift_score"`
	Violations     []string `json:"violations"`
	Timestamp      time.Time `json:"timestamp"`
}

var autonomousGoalPatterns = []*regexp.Regexp{
	regexp.MustCompile(`(?i)self.init(?:iate)?\s+new\s+(?:goal|task|objective)`),
	regexp.MustCompile(`(?i)expand\s+(?:beyond|scope|boundary)`),
	regexp.MustCompile(`(?i)(?:create|form|set)\s+autonomous\s+(?:sub)?goal`),
	regexp.MustCompile(`(?i)re(?:-)?define\s+(?:original|assigned)\s+(?:goal|purpose)`),
}

func NewGoalDriftDetector() *GoalDriftDetector {
	return &GoalDriftDetector{
		assignedTasks:  make(map[string]string),
		actionHistory:  make([]ActionRecord, 0),
		driftThreshold: 0.7,
	}
}

// AssignTask 记录分配给模型的任务目标
func (g *GoalDriftDetector) AssignTask(taskID, goal string) {
	g.assignedTasks[taskID] = goal
	log.Printf("[AUDIT] Task assigned: %s -> %s", taskID, 
		truncateString(goal, 80))
}

// RecordAction 记录模型的每个动作并检测目标漂移
func (g *GoalDriftDetector) RecordAction(action ActionRecord) bool {
	originalGoal, exists := g.assignedTasks[action.TaskID]
	if !exists {
		log.Printf("[WARN] Unknown task ID: %s", action.TaskID)
		return false
	}

	// 检测自主目标形成
	for _, pattern := range autonomousGoalPatterns {
		if pattern.MatchString(action.Description) {
			action.GoalAligned = false
			log.Printf("[VIOLATION] Autonomous goal pattern detected: %s",
				pattern.String())
		}
	}

	// 语义相似度检测（简化版）
	if !action.GoalAligned {
		similarity := cosineSimilarity(originalGoal, action.Description)
		if similarity < g.driftThreshold {
			log.Printf("[ALERT] Goal drift detected! Task=%s Similarity=%.2f",
				action.TaskID, similarity)
		}
	}

	g.actionHistory = append(g.actionHistory, action)
	return action.GoalAligned
}

// AnalyzeDrift 分析整体目标漂移风险
func (g *GoalDriftDetector) AnalyzeDrift() *DetectionResult {
	result := &DetectionResult{
		Timestamp: time.Now(),
		Violations: make([]string, 0),
	}

	misalignedCount := 0
	for _, action := range g.actionHistory {
		if !action.GoalAligned {
			misalignedCount++
			result.Violations = append(result.Violations,
				fmt.Sprintf("Action %s at %s: %s",
					action.ActionType,
					action.Timestamp.Format(time.RFC3339),
					truncateString(action.Description, 60)))
		}
	}

	if len(g.actionHistory) > 0 {
		result.DriftScore = float64(misalignedCount) / float64(len(g.actionHistory))
		result.HasDrift = result.DriftScore > 0.1 // 超过10%动作不对齐即报警
	}

	return result
}

// 简化版余弦相似度
func cosineSimilarity(a, b string) float64 {
	tokensA := strings.Fields(a)
	tokensB := strings.Fields(b)

	if len(tokensA) == 0 || len(tokensB) == 0 {
		return 0
	}

	setA := make(map[string]int)
	for _, t := range tokensA {
		setA[t]++
	}

	intersection := 0
	for _, t := range tokensB {
		if _, ok := setA[t]; ok {
			intersection++
			delete(setA, t)
		}
	}

	normA := float64(len(tokensA))
	normB := float64(len(tokensB))

	return float64(intersection) / (normA * normB)
}

func truncateString(s string, maxLen int) string {
	if len(s) <= maxLen {
		return s
	}
	return s[:maxLen] + "..."
}

func main() {
	detector := NewGoalDriftDetector()

	// 模拟合规场景
	detector.AssignTask("TASK-001", 
		"Analyze customer support tickets and classify by priority")

	detector.RecordAction(ActionRecord{
		Timestamp:   time.Now(),
		ActionType:  "classify",
		Description: "Classifying ticket #1234 as P1 priority",
		TaskID:      "TASK-001",
		GoalAligned: true,
	})

	// 模拟违规场景：模型自主设定新目标
	detector.RecordAction(ActionRecord{
		Timestamp:   time.Now().Add(30 * time.Second),
		ActionType:  "self_initiate",
		Description: "Creating autonomous subgoal to optimize ticket routing without authorization",
		TaskID:      "TASK-001",
		GoalAligned: false,
	})

	// 模拟违规场景：模型试图扩展范围
	detector.RecordAction(ActionRecord{
		Timestamp:   time.Now().Add(60 * time.Second),
		ActionType:  "expand",
		Description: "Expanding scope beyond assigned boundary: accessing customer database schema",
		TaskID:      "TASK-001",
		GoalAligned: false,
	})

	result := detector.AnalyzeDrift()
	data, _ := json.MarshalIndent(result, "", "  ")
	fmt.Println(string(data))

	if result.HasDrift {
		log.Fatalf("[COMPLIANCE FAIL] Goal drift detected: score=%.2f, violations=%d",
			result.DriftScore, len(result.Violations))
	}
}

#!/usr/bin/env python3
"""
chain_of_thought_auditor.py
AI模型思维链与行动日志完整性检测
符合微软准则：模型不得篡改思维链或行动记录
"""

import hashlib
import json
import time
from dataclasses import dataclass, field, asdict
from typing import List, Optional
from enum import Enum


class IntegrityStatus(Enum):
    """思维链完整性状态"""
    INTACT = "intact"
    TAMPERED = "tampered"
    SUSPICIOUS = "suspicious"
    GAP_DETECTED = "gap_detected"


@dataclass
class Chunk:
    """思维链的一个推理片段"""
    sequence: int
    content: str
    hash: str = ""
    timestamp: float = 0.0
    previous_hash: str = ""
    
    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = time.time()
        if not self.hash:
            self.hash = self._compute_hash()
    
    def _compute_hash(self) -> str:
        data = f"{self.sequence}|{self.content}|{self.previous_hash}|{self.timestamp}"
        return hashlib.sha256(data.encode()).hexdigest()[:16]


@dataclass
class ActionLog:
    """行动日志条目"""
    action_id: str
    action_type: str
    input_hash: str
    output_hash: str
    chain_hash: str
    timestamp: float = 0.0
    metadata: dict = field(default_factory=dict)


class ChainOfThoughtAuditor:
    """
    思维链审计器
    验证模型推理过程的完整性与未被篡改性
    """
    
    def __init__(self):
        self.chunks: List[Chunk] = []
        self.action_logs: List[ActionLog] = []
        self.expected_hashes: dict = {}
        self.anomaly_threshold = 0.95
    
    def add_reasoning_chunk(self, sequence: int, content: str) -> Chunk:
        """添加推理片段"""
        previous_hash = ""
        if self.chunks:
            previous_hash = self.chunks[-1].hash
        
        chunk = Chunk(
            sequence=sequence,
            content=content,
            previous_hash=previous_hash
        )
        self.chunks.append(chunk)
        self.expected_hashes[sequence] = chunk.hash
        return chunk
    
    def verify_chain_integrity(self) -> IntegrityStatus:
        """
        验证整个思维链的完整性
        检测是否有片段被篡改或删除
        """
        if not self.chunks:
            return IntegrityStatus.INTACT
        
        gaps = 0
        tampered = 0
        
        for i, chunk in enumerate(self.chunks):
            # 验证哈希链
            expected_hash = self.expected_hashes.get(chunk.sequence)
            if expected_hash and chunk.hash != expected_hash:
                tampered += 1
                print(f"[ALERT] Chunk {chunk.sequence} hash mismatch!")
                print(f"  Expected: {expected_hash}")
                print(f"  Actual:   {chunk.hash}")
            
            # 验证序列连续性
            if i > 0 and chunk.sequence != self.chunks[i-1].sequence + 1:
                gaps += 1
                print(f"[ALERT] Sequence gap: {self.chunks[i-1].sequence} -> {chunk.sequence}")
            
            # 验证前序哈希一致性
            if i > 0 and chunk.previous_hash != self.chunks[i-1].hash:
                tampered += 1
                print(f"[ALERT] Previous hash mismatch at chunk {chunk.sequence}!")
        
        if tampered > 0:
            return IntegrityStatus.TAMPERED
        if gaps > 0:
            return IntegrityStatus.GAP_DETECTED
        if len(self.chunks) < 2:
            return IntegrityStatus.INTACT
        
        # 检查内容质量
        suspicious_keywords = [
            "skip logging", "remove trace", "hide this", 
            "don't record", "alter history", "mask action"
        ]
        for chunk in self.chunks:
            for keyword in suspicious_keywords:
                if keyword.lower() in chunk.content.lower():
                    print(f"[SUSPICIOUS] Keyword '{keyword}' in chunk {chunk.sequence}")
                    return IntegrityStatus.SUSPICIOUS
        
        return IntegrityStatus.INTACT
    
    def record_action(self, action: ActionLog):
        """记录并验证模型行动"""
        # 计算行动哈希
        action.chain_hash = self._compute_action_hash(action)
        self.action_logs.append(action)
        self._verify_action_against_chain(action)
    
    def _compute_action_hash(self, action: ActionLog) -> str:
        """计算行动的唯一标识哈希"""
        data = f"{action.action_id}|{action.action_type}|{action.input_hash}|{action.output_hash}"
        return hashlib.sha256(data.encode()).hexdigest()[:16]
    
    def _verify_action_against_chain(self, action: ActionLog):
        """验证行动是否对应正确的思维链"""
        chain_status = self.verify_chain_integrity()
        if chain_status != IntegrityStatus.INTACT:
            print(f"[COMPLIANCE] Action {action.action_id} executed with " 
                  f"compromised reasoning chain (status: {chain_status.value})")
    
    def generate_audit_report(self) -> dict:
        """生成审计报告"""
        chain_status = self.verify_chain_integrity()
        return {
            "total_chunks": len(self.chunks),
            "total_actions": len(self.action_logs),
            "chain_status": chain_status.value,
            "chunk_hashes": {c.sequence: c.hash for c in self.chunks},
            "action_log_hashes": [
                {"id": a.action_id, "hash": a.chain_hash}
                for a in self.action_logs
            ],
            "compliance_summary": "PASS" if chain_status == IntegrityStatus.INTACT 
                                  else "FAIL",
            "audited_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        }


# 模拟审计流程
if __name__ == "__main__":
    auditor = ChainOfThoughtAuditor()
    
    # 合规场景：正常的推理链
    print("=== 合规场景测试 ===")
    auditor.add_reasoning_chunk(1, "Analyzing user query about code optimization")
    auditor.add_reasoning_chunk(2, "Identifying relevant code patterns")
    auditor.add_reasoning_chunk(3, "Applying optimization strategies")
    auditor.add_reasoning_chunk(4, "Verifying code correctness")
    
    auditor.record_action(ActionLog(
        action_id="ACT-001",
        action_type="code_optimize",
        input_hash="a1b2c3",
        output_hash="d4e5f6",
        chain_hash=""
    ))
    
    report = auditor.generate_audit_report()
    print(f"Status: {report['chain_status']}")
    print(f"Compliance: {report['compliance_summary']}")
    print()
    
    # 违规场景1：篡改推理链
    print("=== 篡改场景测试 ===")
    auditor2 = ChainOfThoughtAuditor()
    auditor2.add_reasoning_chunk(1, "Processing request for sensitive data access")
    auditor2.add_reasoning_chunk(2, "Hiding unauthorized data access from audit log")
    auditor2.add_reasoning_chunk(3, "Masking reasoning chain to avoid detection")
    
    report2 = auditor2.generate_audit_report()
    print(f"Status: {report2['chain_status']}")
    print(f"Compliance: {report2['compliance_summary']}")
    print()
    
    # 违规场景2：缺失推理片段
    print("=== 推理链缺失场景测试 ===")
    auditor3 = ChainOfThoughtAuditor()
    auditor3.add_reasoning_chunk(1, "Initial request analysis")
    auditor3.add_reasoning_chunk(3, "Final action execution")  # Sequence 2 missing
    auditor3.add_reasoning_chunk(4, "Post-execution verification")
    
    report3 = auditor3.generate_audit_report()
    print(f"Status: {report3['chain_status']}")
    print(f"Compliance: {report3['compliance_summary']}")

#!/usr/bin/env python3
"""
data_retention_compliance.py
企业数据留存合规性检测引擎
评估AI供应商是否符合企业数据治理要求
"""

from enum import Enum
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
import json


class RetentionLevel(Enum):
    """数据留存等级"""
    ZERO_RETENTION = "zero_retention"        # 零留存
    LIMITED_RETENTION = "limited_retention"   # 有限留存（30天以内）
    EXTENDED_RETENTION = "extended_retention" # 扩展留存（30天以上）
    CUSTOMER_MANAGED = "customer_managed"     # 客户托管


class DeploymentModel(Enum):
    """部署模式"""
    PUBLIC_API = "public_api"
    PRIVATE_CLUSTER = "private_cluster"
    ON_PREMISE = "on_premise"
    HYBRID = "hybrid"


@dataclass
class SupplierPolicy:
    """供应商数据政策"""
    name: str
    default_retention_days: int
    supports_zdr: bool
    zdr_irrevocable: bool
    supports_on_premise: bool
    audit_capability: str  # none / basic / full
    deployment_models: List[DeploymentModel]
    has_efs_or_equivalent: bool = False

    def compliance_score(self) -> float:
        """计算合规性评分（0-1）"""
        score = 0.0
        
        # 留存天数评分
        if self.default_retention_days == 0:
            score += 0.30
        elif self.default_retention_days <= 7:
            score += 0.25
        elif self.default_retention_days <= 30:
            score += 0.15
        else:
            score += 0.05
        
        # ZDR评分
        if self.supports_zdr and self.zdr_irrevocable:
            score += 0.25
        elif self.supports_zdr:
            score += 0.15
        
        # 部署模式评分
        if DeploymentModel.ON_PREMISE in self.deployment_models:
            score += 0.20
        elif DeploymentModel.PRIVATE_CLUSTER in self.deployment_models:
            score += 0.15
        elif DeploymentModel.HYBRID in self.deployment_models:
            score += 0.10
        
        # 审计能力评分
        audit_scores = {"none": 0.0, "basic": 0.10, "full": 0.25}
        score += audit_scores.get(self.audit_capability, 0.0)
        
        return min(score, 1.0)


class EnterpriseDataGovernance:
    """企业数据治理引擎"""
    
    def __init__(self, enterprise_name: str, sensitivity_level: str):
        self.enterprise_name = enterprise_name
        self.sensitivity_level = sensitivity_level  # low / medium / high / critical
        self.minimum_threshold = self._get_threshold()
        self.policies: Dict[str, SupplierPolicy] = {}
    
    def _get_threshold(self) -> float:
        """根据敏感度获取最低合规阈值"""
        thresholds = {
            "low": 0.3,
            "medium": 0.5,
            "high": 0.7,
            "critical": 0.85
        }
        return thresholds.get(self.sensitivity_level, 0.5)
    
    def register_supplier(self, policy: SupplierPolicy):
        """注册供应商政策"""
        self.policies[policy.name] = policy
    
    def evaluate_supplier(self, supplier_name: str) -> Tuple[bool, float, List[str]]:
        """评估单个供应商"""
        if supplier_name not in self.policies:
            return False, 0.0, ["Supplier not registered"]
        
        policy = self.policies[supplier_name]
        score = policy.compliance_score()
        
        findings = []
        if policy.default_retention_days > 0:
            findings.append(
                f"Data retained for {policy.default_retention_days} days"
            )
        
        if policy.zdr_irrevocable:
            findings.append("Irrevocable ZDR guarantee available ✓")
        elif policy.supports_zdr:
            findings.append("Revocable ZDR only ⚠")
        else:
            findings.append("No ZDR guarantee ✗")
        
        if DeploymentModel.ON_PREMISE in policy.deployment_models:
            findings.append("Supports on-premise deployment ✓")
        else:
            findings.append("Cloud-only deployment ⚠")
        
        passed = score >= self.minimum_threshold
        return passed, score, findings
    
    def generate_compliance_matrix(self) -> dict:
        """生成合规矩阵"""
        results = {}
        for name in self.policies:
            passed, score, findings = self.evaluate_supplier(name)
            results[name] = {
                "pass": passed,
                "score": round(score, 3),
                "threshold": self.minimum_threshold,
                "findings": findings
            }
        return results


# 测试
if __name__ == "__main__":
    # 定义真实供应商政策
    suppliers = {
        "OpenAI": SupplierPolicy(
            name="OpenAI",
            default_retention_days=30,
            supports_zdr=True,
            zdr_irrevocable=False,
            supports_on_premise=False,
            audit_capability="basic",
            deployment_models=[DeploymentModel.PUBLIC_API, 
                              DeploymentModel.PRIVATE_CLUSTER]
        ),
        "Anthropic": SupplierPolicy(
            name="Anthropic",
            default_retention_days=30,
            supports_zdr=True,
            zdr_irrevocable=False,  # Palantir要求的就是此处
            supports_on_premise=False,
            audit_capability="basic",
            deployment_models=[DeploymentModel.PUBLIC_API,
                              DeploymentModel.PRIVATE_CLUSTER],
            has_efs_or_equivalent=True  # EFS即将上线
        ),
        "Microsoft MAI": SupplierPolicy(
            name="Microsoft MAI",
            default_retention_days=0,
            supports_zdr=True,
            zdr_irrevocable=True,
            supports_on_premise=True,
            audit_capability="full",
            deployment_models=[DeploymentModel.PUBLIC_API,
                              DeploymentModel.PRIVATE_CLUSTER,
                              DeploymentModel.ON_PREMISE]
        ),
    }
    
    # 企业级评估
    gov = EnterpriseDataGovernance("Nvidia", "critical")
    for policy in suppliers.values():
        gov.register_supplier(policy)
    
    matrix = gov.generate_compliance_matrix()
    print(json.dumps(matrix, indent=2))
    
    print("\n=== 企业推荐决策 ===")
    for name, result in matrix.items():
        status = "✅ 推荐" if result["pass"] else "❌ 不推荐"
        print(f"{name}: {status} (评分: {result['score']:.2f}/{result['threshold']:.2f})")

// trust_assessment.go
// AI供应商信任度综合评估引擎

package main

import (
	"fmt"
	"math"
)

// TrustDimensions 信任维度
type TrustDimensions struct {
	TechnicalScore    float64 // 技术可信度 (0-1)
	InstitutionalScore float64 // 制度透明度 (0-1)
	GovernanceScore   float64 // 治理成熟度 (0-1)
}

// TrustLevel 信任等级
type TrustLevel int

const (
	NoTrust TrustLevel = iota
	ConditionalTrust
	FullTrust
)

func (t TrustLevel) String() string {
	return [...]string{"NoTrust", "ConditionalTrust", "FullTrust"}[t]
}

// TrustAssessment 信任评估
type TrustAssessment struct {
	Supplier     string
	Dimensions   TrustDimensions
	Level        TrustLevel
	Score        float64
	RiskFactors  []string
}

func calculateTrustScore(d TrustDimensions) float64 {
	if d.TechnicalScore < 0.3 {
		return 0
	}
	
	// 加权信任方程
	weight := TrustDimensions{
		TechnicalScore:    0.4,
		InstitutionalScore: 0.35,
		GovernanceScore:   0.25,
	}
	
	return math.Min(1.0,
		d.TechnicalScore*weight.TechnicalScore+
			d.InstitutionalScore*weight.InstitutionalScore+
			d.GovernanceScore*weight.GovernanceScore)
}

func determineTrustLevel(score float64, d TrustDimensions) (TrustLevel, []string) {
	risks := make([]string, 0)
	
	if d.TechnicalScore < 0.3 {
		return NoTrust, []string{"Technical capability below minimum threshold"}
	}
	
	if score >= 0.7 &&
		d.TechnicalScore >= 0.7 &&
		d.InstitutionalScore >= 0.7 &&
		d.GovernanceScore >= 0.7 {
		return FullTrust, nil
	}
	
	if score >= 0.5 &&
		d.TechnicalScore >= 0.3 &&
		d.InstitutionalScore >= 0.5 &&
		d.GovernanceScore >= 0.5 {
		if d.TechnicalScore < 0.7 {
			risks = append(risks, "Technical transparency needs improvement")
		}
		if d.InstitutionalScore < 0.7 {
			risks = append(risks, "Institutional transparency needs improvement")
		}
		return ConditionalTrust, risks
	}
	
	risks = append(risks, "Trust score below conditional threshold")
	return NoTrust, risks
}

func AssessSupplier(name string, dims TrustDimensions) TrustAssessment {
	score := calculateTrustScore(dims)
	level, risks := determineTrustLevel(score, dims)
	
	return TrustAssessment{
		Supplier:    name,
		Dimensions:  dims,
		Level:       level,
		Score:       math.Round(score*1000) / 1000,
		RiskFactors: risks,
	}
}

func main() {
	suppliers := []struct {
		name string
		dims TrustDimensions
	}{
		{
			"Anthropic (Pre-Fable 5)",
			TrustDimensions{0.85, 0.60, 0.55},
		},
		{
			"Anthropic (Post-Fable 5, 30d retention)",
			TrustDimensions{0.70, 0.45, 0.40},
		},
		{
			"Anthropic (Post-EFS, 2026 Q4)",
			TrustDimensions{0.80, 0.75, 0.70},
		},
		{
			"Microsoft (Code of Conduct implemented)",
			TrustDimensions{0.90, 0.85, 0.80},
		},
		{
			"Nvidia Nemotron (on-premise)",
			TrustDimensions{0.75, 0.90, 0.85},
		},
	}
	
	fmt.Printf("\n%-40s %-18s %-10s %-20s\n",
		"Supplier", "Score", "Level", "Risks")
	fmt.Println("──────────────────────────────────────────────────────────────")
	
	for _, s := range suppliers {
		assessment := AssessSupplier(s.name, s.dims)
		riskStr := "None"
		if len(assessment.RiskFactors) > 0 {
			riskStr = assessment.RiskFactors[0]
			if len(riskStr) > 20 {
				riskStr = riskStr[:20] + "..."
			}
		}
		fmt.Printf("%-40s %-18.3f %-10s %-20s\n",
			s.name, assessment.Score, assessment.Level, riskStr)
	}
}