普贤教程 › 生态包

Web 安全与校验实战

令牌、会话、清洗、校验、缓冲区、错误记录、邮件构造、国际化。这一组包的特点相同：状态显式传递，没有隐藏的全局变量与后台线程，所以可以直接放进无状态的请求处理函数里。

版本aescbc / htmlsanitizer 用 0.2.0：前者加解密返 bytes，后者按结构解析属性。下面每段输出都是实测原文。

一个脚本跑完其余 11 个包

import descrypt
import hashutil
import csrf
import session
import htmlsanitizer
import jsonschema
import outbuf
import errorhandler
import mimebuilder
import cldr
import gettext

var seen = []

def on_err(level, msg):
    seen.append(msg)
    return level

def main():
    var c = rc4_encrypt_str("secret", "attack at dawn")
    print(c[0:16] + "… → " + rc4_decrypt_str("secret", c))

    print(str(hu_crc32("hello")) + " " + str(hu_adler32("hello")) + " " + str(hu_fnv1a_32("hello")))

    var tok = csrf_generate(16)
    print(str(len(tok)) + " " + str(csrf_verify(tok, tok)) + " " + str(csrf_verify(tok, tok + "0")))
    print(csrf_input_tag(tok))

    var store = ss_new({"cookie": "PXSESSION"})
    var sid = ss_start(store)
    ss_set(store, sid, "uid", 1001)
    print(str(ss_get(store, sid, "uid")) + " " + str(ss_has(store, sid)))
    var sid2 = ss_regenerate(store, sid)
    print("regen=" + str(sid != sid2) + " 旧会话=" + str(ss_has(store, sid)))

    var dirty = "<p onclick=\"x()\">hi</p><script>alert(1)</script>"
    print(hs_sanitize(dirty, {"allowed": ["p"]}))
    print(str(hs_is_safe_url("javascript:alert(1)")) + " " + str(hs_is_safe_url("https://example.com/a?b=1")))

    var schema = {"type": "object", "required": ["name", "age"],
                  "properties": {"name": {"type": "string"}, "age": {"type": "integer", "minimum": 0}}}
    print(json_stringify(js_validate({"name": "甲", "age": 20}, schema)))
    print(json_stringify(js_validate({"name": "甲", "age": -1}, schema)))
    print(json_stringify(js_validate({"name": "甲"}, schema)))

    var ob = ob_new()
    ob_start(ob)
    ob_write(ob, "part1|")
    ob_start(ob)
    ob_write(ob, "inner")
    print(str(ob_get_level(ob)) + " " + ob_get_clean(ob))
    ob_write(ob, "part2")
    print(ob_get_clean(ob))

    var eh = eh_new()
    eh_set_handler(eh, on_err)
    eh_trigger(eh, EH_WARNING, "磁盘快满了")
    eh_trigger(eh, EH_NOTICE, "缓存未命中")
    print(str(len(seen)) + " " + seen[0] + " / " + eh_last_error(eh)["message"])

    var msg = mb_new()
    mb_set_from(msg, "a@example.com", "甲")
    mb_set_to(msg, "b@example.com", "乙")
    mb_set_subject(msg, "周报")
    mb_set_body(msg, "本周完成 3 项", "<b>本周完成 3 项</b>")
    var raw = mb_build(msg)
    print(str(len(raw) > 300) + " " + str(raw.contains("Subject:")))

    print(cldr_plural("en", 1) + " " + cldr_plural("ru", 3) + " " + cldr_plural("zh", 5))
    var loc = cldr_locale_new("zh", "CN")
    print(cldr_locale_code(loc) + " " + cldr_format_number(loc, 1234567.891))

    var po = "msgid \"\"\nmsgstr \"\"\n\nmsgid \"Hello\"\nmsgstr \"你好\"\n"
    var cat = gt_parse_po(po)
    print(gt_get_translation(cat, "Hello"))

text

8c42a67de1cff6c7… → attack at dawn
907060870 103547413 1335831723
32 true false
<input type="hidden" name="_csrf_token" value="fc92650475b72abf1984c926daa48582">
1001 true
regen=true 旧会话=false
<p>hi</p>
false true
{"valid":true,"errors":[]}
{"valid":false,"errors":["property 'age': number below minimum: -1 < 0"]}
{"valid":false,"errors":["missing required property: age"]}
2 inner
part1|part2
2 磁盘快满了 / 缓存未命中
true true
one few other
zh-CN 1234567.891
你好

aescbc：CBC / CTR 与 openssl 逐字节一致

import aescbc

def main():
    var key = hex_to_bytes("000102030405060708090a0b0c0d0e0f")
    var iv = hex_to_bytes("101112131415161718191a1b1c1d1e1f")
    var ct = aes_cbc_encrypt(key, iv, bytes("hello puxian"))
    print(bytes_to_hex(ct))                       # 2fade11e57401798d6a521bea402f7b8
    print(bytes_to_str(aes_cbc_decrypt(key, iv, ct)))   # hello puxian
    var nonce = hex_to_bytes("000102030405060708090a0b")
    var c2 = aes_ctr_crypt(key, nonce, bytes("hello puxian"))
    print(bytes_to_hex(c2))                       # 9e0210fb9da0b574c7165abe
    print(str(aes_cbc_decrypt_ok(key, iv, bytes("short"))))

text

2fade11e57401798d6a521bea402f7b8
hello puxian
9e0210fb9da0b574c7165abe
Err(AES-CBC 解密失败：密文长度不是 16 的倍数，或 PKCS#7 填充非法)

参数与返回值：key / iv / data 都可用 bytes、list[int] 或字符串（内部统一转 bytes），key 16/24/32 字节即 AES-128/192/256，iv 固定 16 字节。加密返回 bytes，解密失败返 null——要区分「解密失败」与「明文恰好是空」，用 aes_cbc_decrypt_ok 拿 Err。

装包写准版本pxpkg add aescbc@0.2.0（或 @^0.2.0）；^0.1.0 的上界是 0.2.0 开区间，写它装不到这一版。

descrypt：RC4 可用，DES 是教学版

var c = rc4_encrypt_str("secret", "attack at dawn")   # hex 串
rc4_decrypt_str("secret", c)                           # attack at dawn

RC4 是完整实现（虽然 RC4 本身已不推荐用于安全场景）；DES/3DES 是简化教学实现，包自己的头注释就写明「不应用于安全场景」。要真加密用内建 aes_encrypt + 认证（HMAC），或者 passhash 做口令哈希。

hashutil：13 种非加密哈希

hu_crc32("hello")      # 907060870
hu_adler32("hello")    # 103547413
hu_fnv1a_32("hello")   # 1335831723
hu_murmur3_32("hello", 0)
hu_hamming("karolin", "kathrin")   # 3

这些是校验和与散列，不是加密。做去重键、分片路由、Bloom 过滤时用；存密码请用 bcrypt 或 passhash。注意 hu_fnv1_64 / hu_fnv1a_64 内部按 32 位取模（头注释已声明），64 位精度别指望。

csrf：令牌与恒定时间比较

var tok = csrf_generate(16)        # 16 字节 → 32 位 hex
csrf_verify(tok, tok)              # true
csrf_verify(tok, tok + "0")        # false（长度不同直接 false）
csrf_input_tag(tok)                # <input type="hidden" name="_csrf_token" value="…">
csrf_meta_tag(tok)                 # <meta name="csrf-token" content="…">

比较是恒定时间的（逐字节 XOR 累积），不会因为前缀匹配长度泄露信息。csrf_field_name() / csrf_session_key() 给默认字段名，两端对齐用。

令牌要存 session，别只放 cookie这是双提交校验：cookie 一份、会话一份，服务端比对。session 包给的就是内存会话存储，两者天然配合。

session：纯内存会话

var store = ss_new({"cookie": "PXSESSION"})
var sid = ss_start(store)
ss_set(store, sid, "uid", 1001)
ss_get(store, sid, "uid")          # 1001（同时刷新 _last_access）
ss_regenerate(store, sid)          # 新 id，旧会话被销毁 → 防固定会话攻击
ss_gc(store, 3600)                 # 清理超过 1 小时未访问的

无 IO：所有会话都在 store["_sessions"] 这个 dict 里，进程重启即丢。要跨请求保留，自己把 dict 落盘或写 Redis——这个包只负责状态机，不负责存储。

htmlsanitizer：清洗、URL 校验、转义

import htmlsanitizer

def main():
    var o = {"allowed": ["p", "a", "img"]}
    print(hs_sanitize("<p onclick=\"x()\">hi</p>", o))                  # 事件属性整条去掉
    print(hs_sanitize("<a href=\"/ok\" onclick=\"x()\" title=\"t\">x</a>", o))
    print(hs_sanitize("<a href=\"javascript:alert(1)\" title=\"t\">x</a>", o))   # 危险协议去掉 href
    print(hs_sanitize("<img src=\"x.png\" onerror=\"alert(1)\">", o))
    print(str(hs_is_safe_url("java\tscript:alert(1)")))                 # 制表符绕过
    print(str(hs_is_safe_url("&#106;avascript:alert(1)")))              # 实体编码绕过
    print(str(hs_is_safe_url("https://id.xiusoft.cn/portal")))          # 正常地址不误伤

text

<p>hi</p>
<a href="/ok" title="t">x</a>
<a title="t">x</a>
<img src="x.png">
false
false
true

三条校验都挡住了：onclick / onerror 这类事件属性整条删除（不是删子串，不会留 ="x()" 残渣）；href 为 javascript: 时丢掉 href 保留标签；java\tscript: 与 HTML 实体编码两种绕过都被 hs_is_safe_url 识破，正常 https 地址不误伤。

白名单默认很严，不给 allowed 会连 a / img 一起去掉hs_sanitize(html, {}) 只留最基本的文本与少数标签；要保留链接、图片必须显式写进 allowed。

清洗之外还要转义纯文本输出用 hs_escape（< → <）；把用户内容拼进属性值时，属性值本身也要转义。清洗器负责「留下哪些标签」，转义负责「这段文本不许被解释成标签」，两件事都要做。

jsonschema：类型 + 约束

var schema = {"type": "object", "required": ["name", "age"],
              "properties": {"name": {"type": "string"}, "age": {"type": "integer", "minimum": 0}}}

支持 type / required / enum / minimum / maximum / minLength / maxLength / pattern / items，返回 {"valid": bool, "errors": [...]}，错误信息是英文短句。嵌套 properties 会递归校验。

不支持 $ref / oneOf / allOf这是子集实现。复杂 schema 拆成多次 js_validate，或者用 validator 包的字段规则。

outbuf：输出缓冲

var ob = ob_new()
ob_start(ob)
ob_write(ob, "part1|")
ob_start(ob)              # 嵌套
ob_write(ob, "inner")
ob_get_clean(ob)          # "inner"，层级 -1
ob_write(ob, "part2")     # 回到外层
ob_get_clean(ob)          # "part1|part2"

典型用法：模板渲染时先抓成字符串再决定要不要输出；或者包裹一段可能失败的处理，失败就丢弃缓冲。

errorhandler：手动错误记录

var eh = eh_new()
eh_set_handler(eh, on_err)          # 回调签名 (level, msg)
eh_trigger(eh, EH_WARNING, "磁盘快满了")
eh_last_error(eh)["message"]        # 缓存未命中
eh_get_errors(eh)                   # 全部历史

它不接管运行时错误eh_trigger 是你主动调用的记录器，不会捕获 R1001 这类运行时错误（普贤没有 try/catch）。级别用位掩码：EH_WARNING=2、EH_NOTICE=4，eh_set_level 决定哪些级别触发回调。

mimebuilder：构造 MIME 邮件

var msg = mb_new()
mb_set_from(msg, "a@example.com", "甲")
mb_set_to(msg, "b@example.com", "乙")
mb_set_subject(msg, "周报")
mb_set_body(msg, "本周完成 3 项", "<b>本周完成 3 项</b>")
mb_add_attachment(msg, "report.pdf", pdf_bytes, "application/pdf")
var raw = mb_build(msg)

输出完整的 MIME 文本（含 boundary、multipart/alternative、附件 base64）。只构造不发送——真发信接 smtp 包，把 raw 作为 DATA 传过去。解析方向用 mb_parse。

cldr：复数规则与 locale

cldr_plural("en", 1)    # one
cldr_plural("en", 2)    # other
cldr_plural("ru", 3)    # few
cldr_plural("zh", 5)    # other

复数规则覆盖 en / zh / ja / ko / ru / fr / ar 几个主要语系（CLDR 子集）。cldr_format_number 目前不加千分位（输出 1234567.891），要分组自己按三位插逗号。

gettext：解析 .po

var cat = gt_parse_po(po_text)          # {msgid: msgstr}
gt_get_translation(cat, "Hello")        # 你好
gt_get_plural(cat, "%d files", 2)
gt_build_po(entries)                    # 反向生成 .po
gt_compile_mo(catalog)                  # 编译 .mo（简化版）

适合「翻译文件随程序走」的场景。运行时切换语言：把多个 .po 解析成不同 catalog，按键取。.mo 是简化实现，与 GNU gettext 的二进制格式不完全兼容，别拿它替换系统 msgfmt 产物。