importhmacimporthashlibimportjsondefsign_save_data(data: dict, key: bytes) ->tuple:
""" 对存档数据进行HMAC-SHA256签名。 返回: (payload_json, signature_hex) """payload=json.dumps(data, separators=(",", ":"), ensure_ascii=False)
signature=hmac.new(key, payload.encode("utf-8"), hashlib.sha256).hexdigest()
returnpayload, signature

frompydanticimportBaseModel, Field, validatorclassUnitSaveData(BaseModel):
current_hp: int=Field(..., ge=0)
@validator("current_hp")defvalidate_hp_not_excessive(cls, v, values):
base_hp=values.get("base_hp", 100)
ifv>base_hp*1.5: # 允许50%缓冲（buff等）raiseValueError(f"HP {v} exceeds reasonable maximum {base_hp*1.5}")
returnv

importosdefsanitize_path(path: str) ->str:
"""将绝对路径转换为相对路径或脱敏"""home=os.path.expanduser("~")
ifpath.startswith(home):
returnpath.replace(home, "~")
# 或者只显示最后两级目录parts=os.path.normpath(path).split(os.sep)
iflen(parts) >3:
returnos.path.join("...", *parts[-3:])
returnpath

frompydanticimportBaseModel, field_validatorclassTileMapConfig(BaseModel):
width: int=Field(..., ge=16, le=256)
height: int=Field(..., ge=16, le=256)
@field_validator("width", "height")@classmethoddefcheck_map_size(cls, v):
total_tiles=v*v# 假设正方形简化计算max_tiles=256*256# 65536 tilesiftotal_tiles>max_tiles:
raiseValueError(f"Map too large: {total_tiles} tiles exceeds limit {max_tiles}")
returnv

# v2.0 计划功能classModSandbox:
ALLOWED_APIS= {
"register_unit_template",
"register_weapon_config",
"register_terrain_type",
"register_objective_type",
}
FORBIDDEN_MODULES= {"os", "sys", "subprocess", "socket", "importlib", "builtins", "__import__"}
defexecute_mod_script(self, script_path: str):
restricted_globals= {
"__builtins__": self._safe_builtins(),
"api": self._create_api_proxy(),
}
withopen(script_path) asf:
code=compile(f.read(), script_path, "exec")
exec(code, restricted_globals) # 受限环境

importhashlibimportsubprocessimportfunctoolsfromtypingimportOptionalclassSecureIO:
_derived_key: Optional[bytes] =None_KEY_SALT: Optional[bytes] =None_PBKDF2_ITERATIONS=100_000_KEY_LENGTH=32# 256 bits@classmethoddef_init_salt(cls) ->bytes:
"""生成per-install随机Salt，持久化到本地配置文件。"""ifcls._KEY_SALTisnotNone:
returncls._KEY_SALTimportos, secretssalt_path=os.path.join(os.path.expanduser("~"), ".pycc2", "salt.bin")
os.makedirs(os.path.dirname(salt_path), exist_ok=True)
ifos.path.exists(salt_path):
withopen(salt_path, "rb") asf:
cls._KEY_SALT=f.read()
else:
cls._KEY_SALT=secrets.token_bytes(32)
withopen(salt_path, "wb") asf:
f.write(cls._KEY_SALT)
os.chmod(salt_path, 0o600)
returncls._KEY_SALT@classmethoddef_get_device_uuid(cls) ->str:
"""获取macOS设备的唯一标识符 (ioreg UUID)"""try:
result=subprocess.run(
["ioreg", "-rd1", "-c", "IOPlatformExpertDevice"],
capture_output=True,
text=True,
timeout=5,
)
forlineinresult.stdout.split("\n"):
if"UUID"inlineand'"'inline:
uuid=line.split('"')[-2]
iflen(uuid) >=16:
returnuuidexceptException:
pass# Fallback: 使用machine-id或随机生成importplatformfallback=f"{platform.node()}-{platform.machine()}"returnhashlib.sha256(fallback.encode()).hexdigest()
@classmethoddef_derive_key(cls) ->bytes:
""" 使用PBKDF2从设备UUID派生密钥。 结果缓存在类变量中避免重复计算。 """ifcls._derived_keyisnotNone:
returncls._derived_keydevice_uuid=cls._get_device_uuid().encode("utf-8")
salt=cls._init_salt()
key=hashlib.pbkdf2_hmac(
"sha256", device_uuid, salt, cls._PBKDF2_ITERATIONS, dklen=cls._KEY_LENGTH
)
cls._derived_key=keyreturnkey

importjsonimporthmacimporthashlibimportosimporttempfilefrompathlibimportPathfromdatetimeimportdatetimefromtypingimportAny, DictclassSaveCorruptedError(Exception):
passclassSaveVersionError(Exception):
passclassSecureIO:
SIG_PREFIX="#SIG:"SAVE_VERSION="1.0.0"@classmethoddefwrite_save(cls, save_path: Path, data: Dict[str, Any], *, dev_mode: bool=False) ->bool:
""" 安全写入存档文件。 Args: save_path: 目标文件路径 (.sav) data: 要保存的数据字典 (应符合SaveGameData schema) dev_mode: 开发模式下跳过签名 (用于测试) Returns: True 如果成功 Raises: ValueError: 数据校验失败 IOError: 写入失败 """# Step 1: 序列化为紧凑JSONtry:
payload=json.dumps(
data,
separators=(",", ":"),
ensure_ascii=False,
sort_keys=True,
default=str, # 处理datetime等非标准类型
)
except (TypeError, ValueError) ase:
raiseValueError(f"Serialization failed: {e}")
# Step 2: 生成HMAC签名 (除非dev_mode)ifnotdev_mode:
key=cls._derive_key()
signature=hmac.new(key, payload.encode("utf-8"), hashlib.sha256).hexdigest()
file_content=f"{payload}\n{cls.SIG_PREFIX}{signature}"else:
file_content=f"{payload}\n{cls.SIG_PREFIX}DEV_MODE_NO_SIG"# Step 3: 原子写入save_path=Path(save_path)
save_path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_path=tempfile.mkstemp(dir=save_path.parent, suffix=".tmp")
try:
withos.fdopen(fd, "w", encoding="utf-8") asf:
f.write(file_content)
f.flush()
os.fsync(f.fileno())
os.replace(tmp_path, str(save_path))
returnTrueexceptException:
ifos.path.exists(tmp_path):
os.unlink(tmp_path)
raise

importjsonimporthmacimporthashlibfrompathlibimportPathfromtypingimportDict, AnyclassSecureIO:
@classmethoddefread_save(
cls,
save_path: Path,
*,
dev_mode: bool=False
) ->Dict[str, Any]:
""" 安全读取并验证存档文件。 Args: save_path: 存档文件路径 dev_mode: 开发模式下跳过签名验证 Returns: 解析后的数据字典 Raises: FileNotFoundError: 文件不存在 SaveCorruptedError: 文件损坏或签名无效 SaveVersionError: 版本不兼容 ValueError: JSON解析失败或数据校验失败 """save_path=Path(save_path)
ifnotsave_path.exists():
raiseFileNotFoundError(f"Save file not found: {save_path}")
# Step 1: 读取原始内容try:
raw_content=save_path.read_text(encoding='utf-8')
exceptUnicodeDecodeErrorase:
raiseSaveCorruptedError(f"File encoding error: {e}")
exceptIOErrorase:
raiseSaveCorruptedError(f"Read error: {e}")
ifnotraw_content.strip():
raiseSaveCorruptedError("Save file is empty")
# Step 2: 分离payload和签名last_newline=raw_content.rfind('\n')
iflast_newline==-1:
raiseSaveCorruptedError("Invalid file format: no signature line")
payload=raw_content[:last_newline]
sig_line=raw_content[last_newline+1:].strip()
ifnotsig_line.startswith(cls.SIG_PREFIX):
raiseSaveCorruptedError("Invalid signature format")
stored_sig=sig_line[len(cls.SIG_PREFIX):]
# Step 3: 验证签名 (非dev模式)ifnotdev_mode:
ifstored_sig=="DEV_MODE_NO_SIG":
raiseSaveCorruptedError("Dev-mode save cannot be loaded in normal mode")
key=cls._derive_key()
computed_sig=hmac.new(
key,
payload.encode('utf-8'),
hashlib.sha256
).hexdigest()
ifnothmac.compare_digest(computed_sig, stored_sig):
raiseSaveCorruptedError(
"HMAC verification failed - file may be tampered"
)
else:
print("[WARNING] Dev mode: skipping signature verification")
# Step 4: JSON反序列化try:
data=json.loads(payload)
exceptjson.JSONDecodeErrorase:
raiseSaveCorruptedError(f"JSON parse error at position {e.pos}: {e.msg}")
ifnotisinstance(data, dict):
raiseSaveCorruptedError("Save data must be a JSON object")
# Step 5: 版本检查saved_version=data.get('version', 'unknown')
ifsaved_version!=cls.SAVE_VERSIONandsaved_version!='unknown':
ifnotcls._is_version_compatible(saved_version):
raiseSaveVersionError(
f"Incompatible save version: {saved_version} "f"(expected {cls.SAVE_VERSION})"
)
returndata@classmethoddef_is_version_compatible(cls, version: str) ->bool:
""" 检查版本兼容性。 未来可实现版本迁移逻辑。 """major=int(version.split('.')[0])
current_major=int(cls.SAVE_VERSION.split('.')[0'])
returnmajor<=current_major

classSecureIO:
@classmethoddefverify_save_integrity(cls, save_path: Path) ->tuple[bool, str]:
""" 快速验证存档完整性（不加载全部数据）。 Returns: (is_valid: bool, message: str) """try:
cls.read_save(save_path)
returnTrue, "Valid"exceptSaveCorruptedErrorase:
returnFalse, f"Corrupted: {e}"exceptSaveVersionErrorase:
returnFalse, f"Version error: {e}"exceptExceptionase:
returnFalse, f"Error: {e}"@classmethoddefget_save_metadata(cls, save_path: Path) ->dict:
""" 仅读取存档元数据（轻量级操作）。 Returns: 包含version, timestamp, mission_index等的字典 """data=cls.read_save(save_path)
return {
"version": data.get("version"),
"timestamp": data.get("timestamp"),
"mission_index": data.get("campaign_state", {}).get("mission_index"),
"squad_count": len(data.get("squads", {})),
"file_size": save_path.stat().st_size,
}
@classmethoddefcreate_backup(cls, save_path: Path) ->Path:
""" 创建存档备份 (带时间戳)。 """timestamp=datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path=save_path.with_suffix(f".sav.bak_{timestamp}")
importshutilshutil.copy2(save_path, backup_path)
returnbackup_path@classmethoddeflist_saves(cls, saves_dir: Path) ->list[dict]:
""" 列出目录下所有有效存档及其元数据。 """saves= []
forsave_fileinsorted(saves_dir.glob("*.sav")):
try:
metadata=cls.get_save_metadata(save_file)
metadata["path"] =str(save_file)
metadata["filename"] =save_file.namesaves.append(metadata)
exceptException:
saves.append(
{
"path": str(save_file),
"filename": save_file.name,
"valid": False,
"error": "Cannot read",
}
)
returnsaves

classTileMapConfig(BaseModel):
width: int=Field(..., ge=16, le=256)
height: int=Field(..., ge=16, le=256)
terrain_grid: List[List[int]]
spawn_points: List[SpawnPoint]
objectives: List[Objective]
@model_validator(mode="after")defvalidate_consistency(self):
grid_h=len(self.terrain_grid)
grid_w=len(self.terrain_grid[0]) ifgrid_h>0else0ifgrid_h!=self.heightorgrid_w!=self.width:
raiseValueError(
f"Grid dimensions ({grid_w}x{grid_h}) don't match "f"declared size ({self.width}x{self.height})"
)
forspinself.spawn_points:
ifnot (0<=sp.position_x<self.widthand0<=sp.position_y<self.height):
raiseValueError(f"Spawn point {sp.id} out of bounds")
returnself

classUnitTemplate(BaseModel):
id: str=Field(..., pattern=r"^[a-z][a-z0-9_]*$") # ID格式约束name: str=Field(..., min_length=1, max_length=64)
faction: Literal["allies", "axis"]
unit_type: Literal[
"infantry_rifle",
"infantry_mg",
"infantry_at",
"infantry_commander",
"vehicle_light",
"vehicle_medium",
"support_mortar",
]
staff_count_default: int=Field(default=10, ge=1, le=100)
move_speed: float=Field(default=3.0, ge=0.5, le=20.0)
base_hp: int=Field(..., gt=0, le=1000)
weapons: List[str] =Field(..., min_length=1, max_length=10)
classWeaponConfig(BaseModel):
id: str=Field(..., pattern=r"^[a-z][a-z0-9_]*$")
caliber: str=Field(..., min_length=1, max_length=32)
range_max: int=Field(..., gt=0, le=10000)
range_effective: int=Field(..., gt=0, le="range_max")
damage_base: int=Field(..., gt=0, le=1000)
rof: int=Field(..., gt=0, le=5000)
magazine_size: int=Field(..., gt=0, le=1000)
@model_validator(mode="after")defvalidate_range_logic(self):
ifself.range_effective>self.range_max:
raiseValueError(
f"Effective range ({self.range_effective}) cannot exceed "f"max range ({self.range_max})"
)
returnself

try:
config=EngineConfig.model_validate(toml.load("engine.toml"))
exceptValidationErrorase:
logger.warning(f"Config validation error: {e}")
config=EngineConfig() # 回退到默认值logger.info("Using default configuration")

importargparsedefparse_args():
parser=argparse.ArgumentParser(description="PyCC2 - WWII Tactical Combat Game")
parser.add_argument(
"--map", type=str, help="Path to map JSON file", default="data/maps/tutorial.json"
)
parser.add_argument("--save", type=str, help="Path to save file to load", default=None)
parser.add_argument("--debug", action="store_true", help="Enable debug mode")
parser.add_argument(
"--dev-mode", action="store_true", help="Developer mode (skip security checks)"
)
parser.add_argument(
"--resolution", type=str, help="Window resolution (WIDTHxHEIGHT)", default="1280x720"
)
args=parser.parse_args()
# 额外校验分辨率格式ifargs.resolution:
try:
w, h=map(int, args.resolution.split("x"))
assert640<=w<=4096and480<=h<=2160except (ValueError, AssertionError):
parser.error(f"Invalid resolution: {args.resolution}")
returnargs

classInputHandler:
MAX_CLICK_RATE=30# Hz (防连点)MIN_DRAG_DISTANCE=5# px (防误触)defhandle_event(self, event):
ifevent.type==pygame.MOUSEBUTTONDOWN:
x, y=event.pos# 坐标范围检查ifnot (0<=x<SCREEN_WIDTHand0<=y<SCREEN_HEIGHT):
return# 忽略超出屏幕的坐标# 防抖动ifevent.button==1:
now=pygame.time.get_ticks()
ifnow-self.last_click_time<1000//self.MAX_CLICK_RATE:
return# 过滤过快的点击self.last_click_time=nowelifevent.type==pygame.KEYDOWN:
key=event.key# 只接受已知键位VALID_KEYS= {
K_UP,
K_DOWN,
K_LEFT,
K_RIGHT,
K_w,
K_a,
K_s,
K_d,
K_SPACE,
K_ESCAPE,
K_RETURN,
K_F11,
K_HOME,
K_END,
K_f,
K_m,
}
ifkeynotinVALID_KEYSandnotself.debug_mode:
return# 忽略未知按键

MOD_ALLOWED_APIS= {
# === 注册类API (安全, 只写) ==="register_unit_template", # 注册新单位模板"register_weapon_config", # 注册新武器配置"register_terrain_type", # 注册自定义地形属性"register_objective_type", # 注册目标类型"register_mission", # 注册自定义任务# === 查询类API (只读) ==="get_terrain_at", # 查询地形类型"get_distance", # 计算两点距离"get_units_in_radius", # 获取范围内单位列表"get_game_tick", # 获取当前tick"get_difficulty", # 获取难度设置# === 事件订阅 (受控) ==="on_unit_killed", # 单位死亡回调"on_objective_captured", # 目标占领回调"on_mission_start", # 任务开始回调"on_mission_end", # 任务结束回调# === UI扩展 (受限) ==="add_context_menu_item", # 添加右键菜单项"show_notification", # 显示通知横幅"log_to_console", # 输出到控制台 (限速)
}

MOD_FORBIDDEN_MODULES= {
# === 系统交互 ==="os", # 文件系统/进程操作"sys", # 系统参数/路径"subprocess", # 子进程执行"shutil", # 高级文件操作# === 网络 ==="socket", # 网络通信"urllib", # HTTP请求"requests", # HTTP客户端"http", # HTTP服务器/客户端"ftplib", # FTP"telnetlib", # Telnet# === 代码执行 ==="importlib", # 动态导入"importlib.util",
"__import__", # 内置导入"compile", # 代码编译"exec", # 代码执行"eval", # 表达式求值"types", # 动态类型创建# === 序列化安全 ==="pickle", # 反序列化RCE风险"marshal", # 二进制码执行"shelve", # 持久化对象"yaml", # YAML可执行任意Python (!python/tag)# === 其他危险 ==="ctypes", # C库调用"multiprocessing", # 多进程逃逸"threading", # 线程 (可能导致死锁/竞态)"signal", # 信号处理"pty", # 终端模拟
}

importbuiltinsimportsysclassSafeBuiltins(dict):
"""安全的内置命名空间"""_ALLOWED_BUILTINS= {
"True",
"False",
"None",
"int",
"float",
"str",
"bool",
"list",
"dict",
"tuple",
"set",
"len",
"range",
"enumerate",
"zip",
"map",
"filter",
"min",
"max",
"sum",
"abs",
"round",
"print", # 允许但重定向到安全logger"isinstance",
"issubclass",
"type",
"hasattr",
"getattr",
"setattr", # 受限版"Exception",
"ValueError",
"TypeError",
"RuntimeError",
"__name__",
"__doc__",
}
def__missing__(self, key):
ifkeynotinself._ALLOWED_BUILTINS:
raiseSecurityError(
f"Forbidden builtin: '{key}'. This operation is not allowed in mod sandbox."
)
returnbuiltins.__dict__[key]

importresourceimporttimeimporttracemallocclassResourceMonitor:
MAX_CPU_SECONDS=5.0MAX_MEMORY_MB=100MAX_INSTRUCTIONS=10_000def__init__(self):
self.start_time=Noneself.instruction_count=0tracemalloc.start()
def__enter__(self):
self.start_time=time.monotonic()
self.instruction_count=0returnselfdef__exit__(self, exc_type, exc_val, exc_tb):
tracemalloc.stop()
returnFalse# 不吞掉异常defcheck_cpu(self):
elapsed=time.monotonic() -self.start_timeifelapsed>self.MAX_CPU_SECONDS:
raiseTimeoutError(
f"Mod execution exceeded CPU time limit ({elapsed:.1f}s > {self.MAX_CPU_SECONDS}s)"
)
defcheck_memory(self):
current, peak=tracemalloc.get_traced_memory()
current_mb=current/ (1024*1024)
ifcurrent_mb>self.MAX_MEMORY_MB:
raiseMemoryError(
f"Mod memory usage exceeded limit ({current_mb:.1f}MB > {self.MAX_MEMORY_MB}MB)"
)
defincrement_instruction(self):
self.instruction_count+=1ifself.instruction_count%1000==0: # 每1000条指令检查一次self.check_cpu()
self.check_memory()
ifself.instruction_count>self.MAX_INSTRUCTIONS:
raiseLoopLimitExceeded(
f"Instruction count exceeded limit "f"({self.instruction_count} > {self.MAX_INSTRUCTIONS})"
)

importosdefis_dev_mode() ->bool:
"""检查是否处于开发者模式"""returnos.environ.get("PYCC2_DEV_MODE", "").lower() in ("1", "true", "yes") orgetattr(
get_engine_config().debug, "dev_mode", False
)
# 在关键安全位置使用ifnotis_dev_mode():
secure_io.verify_signature(save_data)
else:
logger.warning("⚠️ DEV MODE: Security checks disabled")

MIGRATORS= {
("1.0", "1.1"): migrate_v1_to_v1_1,
("1.1", "1.2"): migrate_v1_1_to_v1_2,
}
defmigrate_save(data: dict, from_ver: str, to_ver: str) ->dict:
key= (from_ver, to_ver)
ifkeynotinMIGRATORS:
raiseSaveVersionError(f"No migrator for {from_ver} → {to_ver}")
returnMIGRATORS[key](data)

importtracebackimportosdefsafe_format_exception(e: Exception) ->str:
"""格式化异常信息, 脱敏敏感路径"""msg=str(e)
# 脱敏绝对路径home=os.path.expanduser("~")
msg=msg.replace(home, "~")
msg=msg.replace(os.getcwd(), ".")
# 限制长度 (防止信息泄漏过多)iflen(msg) >500:
msg=msg[:500] +"... (truncated)"returnmsg# 在顶层异常处理器中使用try:
risky_operation()
exceptExceptionase:
logger.error(f"Operation failed: {safe_format_exception(e)}")
ifnotis_dev_mode():
user_message="An error occurred. Please check logs."else:
user_message=str(e) # 开发模式显示详细信息