From f07716c9fe8b520c05ccf8aaf4c5cc1abd0cea05 Mon Sep 17 00:00:00 2001 From: bot_dev2 Date: Tue, 11 Aug 2026 07:02:35 +0800 Subject: [PATCH] =?UTF-8?q?feat(#95):=20=E5=AE=89=E5=85=A8=E6=A8=A1?= =?UTF-8?q?=E5=9D=97=20-=20RBAC=E8=A7=92=E8=89=B2=E6=9D=83=E9=99=90+JWT?= =?UTF-8?q?=E8=AE=A4=E8=AF=81+AES-256-GCM=E6=95=B0=E6=8D=AE=E5=8A=A0?= =?UTF-8?q?=E5=AF=86+=E5=AE=89=E5=85=A8=E5=AE=A1=E8=AE=A1=E5=93=88?= =?UTF-8?q?=E5=B8=8C=E9=93=BE+=E5=AE=89=E5=85=A8=E4=B8=AD=E9=97=B4?= =?UTF-8?q?=E4=BB=B6(21=E4=B8=AA=E5=8D=95=E5=85=83=E6=B5=8B=E8=AF=95)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + src/security/__init__.py | 15 ++ .../__pycache__/__init__.cpython-312.pyc | Bin 0 -> 721 bytes .../__pycache__/audit.cpython-312.pyc | Bin 0 -> 5053 bytes src/security/__pycache__/auth.cpython-312.pyc | Bin 0 -> 10628 bytes .../__pycache__/encryption.cpython-312.pyc | Bin 0 -> 4376 bytes .../__pycache__/middleware.cpython-312.pyc | Bin 0 -> 5762 bytes src/security/audit.py | 76 +++++++ src/security/auth.py | 188 ++++++++++++++++++ src/security/encryption.py | 62 ++++++ src/security/middleware.py | 91 +++++++++ tests/security/test_security.py | 186 +++++++++++++++++ 12 files changed, 619 insertions(+) create mode 100644 .gitignore create mode 100644 src/security/__init__.py create mode 100644 src/security/__pycache__/__init__.cpython-312.pyc create mode 100644 src/security/__pycache__/audit.cpython-312.pyc create mode 100644 src/security/__pycache__/auth.cpython-312.pyc create mode 100644 src/security/__pycache__/encryption.cpython-312.pyc create mode 100644 src/security/__pycache__/middleware.cpython-312.pyc create mode 100644 src/security/audit.py create mode 100644 src/security/auth.py create mode 100644 src/security/encryption.py create mode 100644 src/security/middleware.py create mode 100644 tests/security/test_security.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 00000000..2c837d1c --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +tests/security/__pycache__/ diff --git a/src/security/__init__.py b/src/security/__init__.py new file mode 100644 index 00000000..c4564b72 --- /dev/null +++ b/src/security/__init__.py @@ -0,0 +1,15 @@ +"""安全模块：RBAC 权限 + JWT 认证 + 数据加密 + 安全审计""" + +from .auth import ( + Role, Permission, TokenPair, UserToken, + create_tokens, verify_token, require_role, require_permission, +) +from .encryption import DataEncryptor, hash_password, verify_password +from .audit import AuditLogger, AuditEntry + +__all__ = [ + "Role", "Permission", "TokenPair", "UserToken", + "create_tokens", "verify_token", "require_role", "require_permission", + "DataEncryptor", "hash_password", "verify_password", + "AuditLogger", "AuditEntry", +] diff --git a/src/security/__pycache__/__init__.cpython-312.pyc b/src/security/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..fc371ef0f77307d9c0931cfb86b9cffb417ae018 GIT binary patch literal 721 zcma)(O=}b}7{`-&*_nNrZtX!l?sZz~d;t+^T@SudN<|4m7pkQVUtsMmDXW3U!b}UF6a#szQCvOlq`_>a>9x z)I;7ZXC+ItiJEj7Ekn=+k^Kgu6J^DBinb~-96T95IvgGLhX?!LKD^pKx3Sq89XvdK z`MkB(x_Eu3H6FYjAMGhIdhverY%qNKdU*6$$@#_Mpg$h;kD+q%U1@D+36Zre!f4ES zoTak5lielhRv0r`y~YWfN$GVM2?fDIZ*tkVPgs1roJ-kYsIp57#B%8ne775>w^^GEm#pSHqB}~?4qMZs>`W`b?@66J$ zH7Xj8hO1H4(BH-C8VwCkV@ac_aMfRt4#rW|!B|>hA$tF7DQjf@1U0FD?4=#*hW}R4 zCBh`Z_#K$2qTuIdAZQ*il4m@)l?hx@jCa*g**Xo=uuCXO#X2v!AT;2t6L8WgSS-o_ z<2a24#;5aAx@Tw>6$v@ZPAO0Qi+ok#y9puu3@%MA6B<*XFwvO00A7HfK>G_={bAX4 KW6xnH)$v~=YSvHy literal 0 HcmV?d00001 diff --git a/src/security/__pycache__/audit.cpython-312.pyc b/src/security/__pycache__/audit.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..62a3f0c38276fde0f361593ee8829038f0cc5ce7 GIT binary patch literal 5053 zcmcIodvFuy5&uqir-vn5#+HmB1RFvi;$ia)dDtZQMIPX!b=oFM$jzZU8`*j|cc&0J z0vsS6xiN8QlCcSCu``niu2UdsW5qP#e-O+j$YFDt4d3dA3Oui14ULp}Mkr>GoV1g#E ziNQJ>V7)9YIWI>`-pkX{>^0NU;eXYZ;{(@@58N1jXZ+2BW2fI8 z?|naYruW+Mld02hU+H{h{OJ4FPP{jE>h-Zx2U9&KuAUeef9v7fBQ6f;C+Hjn8{60=_Oiv z_k^^jZ#s9knOkq=a(j(>6B%UiO1MbGFX=qe^*Xy#erb?QBFF8AE45&Ir6RY7)yf@V zO^5~sVV9`M%Br9k5?kb;9MV>)QB{+Jm8#;aRApa8@oUja5zkoB9@VW?0Wr8k5;t`e zPo29W)9Hpq)IuPg!uq4YdW#KSi#Wib9!gU6)X)UcunZE{rA9dbyGD6$}` zKEL0?DL9$Vwujqwu2n^CkRri$72e@gInb;#l8TpGUapiPqs=FV!Xdv;40O112_iw%Anm4k%=Ax^%^O3kO!PQdh}g~W)EJ++oKh}i%n11|7JmM=;a1WgYIdV|L3m%%`_ zor)|<{!oj~Z4LX{bY2KRF~}%OfmKCAO)EC&nm^>%1R<^0hCC`Vy>}rysz@hEE?*hv zy7Mo)7xq4X==p^E?l^ZakKUax*D&!OVQ8o;K(fSsmztXVQ^#F&^V#>Ti9A2WZ*4yE_D+oIvl=+jEWg-pvRW0uzZ={?xW$h#E{ zYsMpS86eA)t#f9uKr=H#%pzn@JNpS+i@*DJ%#xWKo7J7-+70(o0MX5I$QPDm-3Bg9^0&w;_)xLET@FbeE7i2`T~g=J zh3Gu>9;y|VC{I5I-m-ykONTSZLMxzS8L+8cKtLJu=k$1A^Cs7A{IvSx>NnQ+Rv)VF zFB&OcGFn`Dp}6v_KVEm)wXnbWWc}Ht=YRCG^Upn>^sKtcGVaQF{aD%3qk9hTIVPV7 z4~7#J4~ z7MI@2Ck3TF-d}ict0>KbHS=rMu;FNcO6!fAv@lrJw4p11Tx>K+wgj5LfAptRhlP zQ+ED^h^$0A=H#k7?OF|->Q_KI$v1|=Uv|&yTXJ;O;Z+IuJ@JNHEGXhP_QLb_1(y~r zOSr2BO8Pb*TXSOL;Kq}3V#R~M6#Gl!4d>leN&B4h_B$!hrZmregZG_$F|ndL!?P(l zHP8KNp7n$4Pi`2>vfTflusGfE{9A6=SKRxLZ_T8n`tN^rZ-W2*_*Wl*!yGIDbNR?s zQ_X#8=8zV-YE9gi%Q&R#T(uVNoRvfBa@9Jxa~=*UwXUoQ!7NiI}8a-)Km-=I7Q^r?)S*#Zs2lxS><$(vG*1mK7$m~SZ$Awf?B(2CKE zZb?Tu%4QsS6o`tzh$dw?566qL;%|-$zMY~!)KQviR=FcMySfj^ly#NN?^}Pg`fzok z#1r3QSeLUj&V40MEbMu?>*YS_f^+euxeG2kOM3Qn?dcPbwjFLe5*%H)c4Xn& zlN%EYA51u_Mx6~K&W42Zk*_xYY3m=hUT{7QZzPLdy;X;*y6PtEq-frRgA}@M6;ngG zt)jaCyI91YTg2Baw@g!XjP$6lSP7Nssnl!{RK_DTrQ<@A&IaU=A>?~uhTQ1SA_y=B z!Jm;VW`jT4`8YCTNCP>I6LK2#u?gp*^VUT)>hqWs4-D#DV>l#xSOvqUDdKS$z8`_! z@QAcV0}ooGsS2&pe1+CEnr^2^t2E&_ z&eTzeK>aMe{(8(}w`aG!BODH78mvF0d2GsZd~F316tfPJk-|R6tH8L4Qj6pxaq=W*;+NjETKIVrMWk}wS9Qr{E6z%PV;y`+(!FGYmg{Sc z@(E!Q+mj3IyzAzChF@%CMmVXbQ28kom=HEcAXkKhP(i&=sKqS7+Bzf^NHCDp?b**h zvZ`>{O;bV)Jm7;zv>kE-1wE0G_R-x>&j;aCf{}Y}QoO(iAWKv;bl;u#HFw^-_jm7m&;OQ}XQUu?f3?T|!UBr=Ek3BptT6QSe=!s_MD3?| z%0=-s&jjdp#>LRcYXTaVhUBb^B{}EfNUn8hNv?D0NUnG3Np5f%AZG){c9YB0Zg!c; zI9wpF-QuzkUK_Bs=ezRT3tR>5g|0%HVyFg+*PW(#y-*}ruCt0Nd9h&g zuTeiDn=_P9A{1Yzl@e&PE?cK^N*n)r^*Ngh#s0^#PbvqcH7cLYq4e?WlgdHqW|hz8 zQ2J!{N#&q)tIB8dqq-dFpU#%6+!sneznt?(bX1e`mTl_T**yK0`Q)S3o<7#~2yYT9 zcr$(+znITE&AOKHOLz<9%gMS~VcqhD71uS&$gY)wiIf*WdEs@?I84`-jB-_`Ym1=P z_L*8IU&a^1$gB8r-oe{XbFS3}ieGA=)PMXVdVIpWZz;F%ksaH@af&xYwLD5 zHan^vd-m4vt8Lx057+~Z^#|*bsjENGxV_$}ylGiC;4LJwsC$J(t9rQ+eG7sdzr zKD~SCX;JuQ*e42Zu_GWr_snyrAh!D?$=4BdOn&do!?#DrFOC==d~nIJ!y|=i_cl5v zZ{Pgn!mWwE*CsB$`!M?Y_}TX-ULAe#{)eAj7(tmI;)FYC+2jcZJ3^ii3?wx>8QH*l zLLP6xBS|uM#3uxJXv5YA!|hIutl#bl1U!cWf~;>038KR7>BPRCfH%E@Bum>KJ-rwn zhbVz^(SRuoPXmT9fFByb4%VgZ=A2rY-3Om0a~{6k7nJoK@L`@%hbU{0`-Cn5a$Y#@ z^9s#Qove4egPwN5?Us#hcY6mP4j^xKyT2Ls1kyDenS5EV&b_Qvre4-6Gk=YmMhXA& zRCTDmvsx57JEZEv9U*sjyW8F6fv;NI?g@I@gmxhqS}S!+A)&om61~-u;0=quPZOj_v?*a)$_Lt2CT{h7fE%e};GtP*n?v zssl9DyNtZSW#Wx4bGON9mW_KeO3-5pOMvT1)y{XX~Z+UDAZdfB9`gL_|nZJlgRKOEfG*osxPt+i0GchA07*}SdsnP(cC z8QlPq-rClEd!9)Tl73LCtS!&AwASx-x74?`LiZNgSi8Tjv6VCxOHd}VAPbRO zq>?4TNVpQPhK1$PT?#2$8hs{*vsFZQ=RDe1M)#x&mYi%-2E?&?mhwJt=xeCZYaBMHhEoX>!?UzaUaIi6Abwopz@^EAF_3s|A#_4%BvtUc_Jgqp|Xg0L8nJVylg z;hM*Uptpk;WUVNLL|-QwCO#uRpiW6Rfqc!!Q_YxZOTx6}PFvEnBhD(8Gi!D*ARx1I0GJ^NOS1EmoeLAi zGd)^Z449K%I)pPyl0O|zsLz;3s0iHy`K-B+=`9bjEg+sIAOyu#z|We9^oKKI71n4v zJ-mdgbAg7S`8#hrHpwODhBvhoCBr9p24z&(GlIAGb1A#t~Mo2)p53( z=#>}xWKvoi=xH>t*`U$QfVhTQ0H=CU>=466&_eNG)f33*(Hj$WBDx6($h>&?}eJIXb?-%0$@Ig4v^Cjjs=cG z4q^;e;BIKu1%Oym2X%VV%ETMMQ|r2ofOUrbfN!lG$AqA)0sWJ8h<9!uFYA#(@F!p_ zv*;*6C#}42#1jsL+($fKa3|e8#q-*uY|1)P$(w6q$iqj`(_bea6ruc(&rP9BkNLFg zSS3ZBpd|=)BQ*2}ot&GFQ!Ugr8W-g$s8W0iuAEp41iUdT34tRri!#nuv7;Wzot3<2 zL1uQUX4mY7airxyK<}(Y=ezs6FV|gZy4aMkuK1O8+sLxfWn=5NCDw0?ui6&ZZzF4q z($u32h-hH#LMSW-W%h7KN5H9{t2|b>J12kuB|<%s4Vk$nA0qC+;pSlP6Ig}>E2Wh{ z&QSlEm;d}TDNDigyHiHX^GzQ)tL}8)b2dhI#S2y>*~S@`=C;#{0%Fhk6%gaqFIPa^ z#HdOL4wlQRUAMt$lnq4H_IiAxh^i=?JzlRMN$wEQz&uenA_~&c^rQB$;1MBIF_Kz$ zsJl~mm~TO7@C8G%Q8>{F5uD`q1)H7PITI*0!ssH}jM#!?&Llm9Wk{ZdpY#Y2n54qu zBwL&+vL{)4%3c<28qg%!vXs3PxqED>GC$a9mOi5oAo?r=iXe9K46nJ&^fFlt{SvHX zc=iUThUD}46g7xfjrngLnfiN2Zew~iyw-1kZ<>=LnN7_XN!ZJVFn%M+Z&tZ%9-+Ue zcZ3bW>MF^PP&_B8TOB`|@|>j3Q7;$K_>1RM!ee&MBAhypY7U#`FZpKh<+^5>BRdvZ z1G^Z&5RE6$CTl<`fv4BV^kzUuqa<{CNSrOnbd}RYJhaUErH-J?@!|GPN!F&_J?rm) zxST!O?(xc+pwK1j0AY>>e1~PNbkqYoBM1b1ZGse%wVuvSA;?2Wuj}-52Rb}FG!eW2 zAhPx-M516JdP!)*$Xb%~Bwt$)Vn?uw&Tt4Y-Pa}|KsrcL!U)cdjl^~+ehz-pH6Vav z#fxM66M3tnJ09rm7eYflgFUeoqpo`;Pmh%}BuW~RC5=ga)BQ}{%2EHllG?G7or#j2 z$&#j|e%BPI$=61APU)%Q1w%UrcMg9wXrXyYCw<=PUaw2Wp143~qVl zso3rxZ@j(r*4CtP+nBL2VQfqqcSX0SSo7IEC-+=vj%`TVR^DSPQzpy#$NL{2*qJmf ziKqRfH|-WdaEd}`Bo??8yv7i0S-~fyJvymRlWg4fb!gqI>19z@E`h~H*h|zKnlIAK zUR@|-6Mk~wlRGlRJ728Vx=_Xz{22|;j*!u74DIKnj9%*vR4AkOSsm^*7(}THAWcG$#W6oaY|es!f>A?|6-NP z=IWP3j68=gJXQx7oGVc#SUE*Uj4f0Jm5lG=xI~m`yo`3kpn-f2JHK+KKAAB%A@KU-M zcRosMrFbHC_2ECBQzAoI1AZ!O&9oT5{lm!*&rV#v;u1<`k;+YNHbdIS*MRml3o9eg+8bbt zAPTVLpo`9(j?cdhwJ5Np18a;@Q;de&m|C&=onu#zjaDRA)Fkv3efojQ;f5(KwZJjp zkI}KpSZUn0X3VxBVFT;`3P1(~S10;U3~R6CUCg^=9a(Z$_o3xIOVZjfW^GPbo0HbP z05~}E{etqCE?H37r=8JK=KL{JS;AB{d}73Y&$RBor6j&!ebTZauHSIqQh46q@4s9& zqE9Y-G-+8MZTLu6Fdz(fCG0D1Ja%)_wN2NbxKr}e;`hpawYq`ybS5pU;`&u7Q(?6E zPcw@sv;FrJ4R6Fz*CZ`#M`7HWzfNhO?5{Is%3S=p)qjz2%XX@G`*P}Et4wta^|R%M z?N4cbR%PD4S@W~a9PmG940Ve&KR21{N;E$&;ehw%2Dk`51nl;ba3ky1a)OLZ@C*}=;ZG^>5Ui^~u zI2<8s_XQpsCi(^(N@xmaStKTPC4aD(lh%I5i^f8pLKK7?eWPzYqB^ z5Sa~!VbjGSeN5II6;2TMEOW;Z8c3+>G>W*}&Au*e&1wW43dp8R{0tA8U`LnuJobnp zIRivO?a3Of5)I@*BmoC>O3W>ZFF}(N*!BvLCAw-yC8)42YS;ga-2m?_<#_xdF%$)v2 zVEc_8En{44)+a+9i*=v#Mvm92QbF{ai7e)m4(jt%iB*gbA`26;~kFgpD}yIP%03<*9Ck z?Cpq8A@3rmA~`O_oc36X6MGr@_rp)>0urSjSc(C?4+V8PI$!9~Na zk-}ubnp9rFecQ4zTV=vlIbyrfef?WW+Y>Ve&a9oHI8Li*UM|pbAPfO!+#%vVFhWC2 z#6@VH#`SN3sI-ncWQWKzXJ;5;vN4XjmJ)r=7Gb<_J=j9AR5u(~G|Of^F=)$rqGDo}gDC^9yHGtVe8v-NmsklG7du6&&U;ZRFtXLn8QYGJ<%K$Ou!o0T>AV zb;ZN>gswbgEglFatjpv2vvZ20EQMgq=6vPzp>HPP z+rS{`X^gzi;sh{#)Digj)8<~73Oo!A*FaU~QUz~kzJGG!)~WILUm5RzNl{GYzGnRL z$oQF)lkeUc?;D!DIx_yknOUcm_$r8m>1jOSc$LPNW!>O>=ehkE%j~22m?Lmqa$2ui zP+B;q*$abZ5-uiq0(yD`m{*zH7t8_@0EMV2>v^*P0wF9j-_cb+dz=<4a(ciB^R=7L z&%stj^F}6oJw-EAza~QY0NCE5zl$@>HveHQ$ot9Ne;ywl%$?SZy~C|rt|q}fTNqI+ zeN7ey?W7!V9S6sn6aDW@UOhK?^X-XOUV8BUWw3bE8O5xhLJlnujQKU>hk$?w1KpI_ zBjKP|=J2#oQq4FTRg_TICE`|A>{*?A`&RX2pu_74NL#BgZp=QV8-EYRmRvyo1g2i( zh}mMtlSNg1b#Uy~7rAdQk304x?R(?-d(-D|F%A?M%3~`>mfdt-bKb1JRvmY4j&I%@ zcRZW4?~CW}`>3d7*cLCUjM)=Km2q<=(dKJ#M~AY9;5-iQ=abl7_MwMr&|RYbay+6mnA*D!-)9qKLsN zuf$~$8;}sL1UM%{WDws5o}5~KBW>K~TnWT?pgM+4FwlWX`vYs>N~UaJwb5PwS9Uvo z=JxnoXUETsJbZp+;@sKscVAcTTmQf97UzoC?SBCCzq8wZZ5?%9<7L|5n2*4<1}qfu zU{1;&h>7&M;)rIDIk$kN;7~ymVZ~*+7S7Lmw7|2csR&z_xuI8s2aI=Bp4d2-Yvbq^ zyl_a@1-92Lz5#8+c%Yz|yfWvT9pZCPG;8Y7Q32G`(|8xRi=qKCBh(={Vng`zL)qN) z8a+hQ)VV@*KtIrQ(c*S!;iMI-XJn>oolJMmIyyHz*(LQsHqo}_2ZdR`;X zR%b?e4_+wYLEa=Par}w>zdZO0`cCbK4evMn^5CB(7HUq}7tphS2_1s?4ImJTWPnM0 z6H9*pMAnh3{a_n0F_|OB{}Sp9_(_K&4O7J0dmjC}zu|{4j)bfCWkB`OBP)~k9aonR zKRXf{tKOEV-uBZ+e_dVw@;774`x@f*9Uock1lQJ$Zn*vUt;ZAi>B%SdCpYX*IQA#) z2jck$<|5dif3KrT8fX&VU$NJ&Vt>K{5fOi8Vc00ji2hT>Fajn$jxYPzTi9=2Ak$@l z47jQMERc!chd$)o9dHZZ`&ry7{OoH*eA7*QTL!(1fL``2dclHMLA}f!?hXl(0+CNb zLAsn=D=4M7AaF6zhOJq=Kqf(FP6N`7ZiRi=acHAdU|gzjUC8Z3(tzYgK;TQBh0GV- zY@nPLB?iVplsy^Yh;JluZ+4{bU=y;{dojb2F)>sGql#iSzYv4ifecJK;Oz_ufqYMq z`WGT!%Un_&f=P&B4B)Cx(UjQPLzjh#h%>qk+rYIsp%B%o?&?0J))xzEvj-eIrrNsouH~`v@C7!+c3b+P{7PsjC9#R(EvX~0W-Cj;tXe-PBsl1 zhF2x?m#0|$+1)314{l273&-?j34K}2IOf=xaBL(6*TJ%g`XMlPU+aDJ&8?wrZmiAXw4MrAf#7Y4>yf1UY}UJKE7x}vf#0) z67xnTPUXi7H%wDznwc)6=>~dQ&(P~;j5K_XVTOVX_UMf1zZo69ot~nQLTJO7rj!C3 zsblFyQ!opQ(ly9g40PGdf_%DYs*>8;Ku>8b^jGOr+0rz#adVp4)jCb8W*#xo+Udnw SI)A#*Oy|#(<&R2&XI(^x_63^ zD>n{98ap8|6sVmHEoSoN43hv;GEhSjn(2T2$3mTvb8%)e6U&&tpm8(J@K@j7=_E`T z(oA>k-PP{feY^X9zxUq#!|AjmDBE}UhJ7}Kz9B!%!WtcF?jUrAk%at6q9i&(MQJ}x zkvbHV4=eJ2^5+hlU(~?uN9kBQvl1s85VEyGZ zdK^iP14t@AMH@Q(&J?o($@Xf5Iaq7)+xDP?o*uN1*@Jp0FQ5E7$jux(GxPdK`Bz8t zr++_v;f?&z+tU}{nK|}$eqcC%{@wh`=fCOSZ@1Tawu-9Q8Vlf5LXG3=m)`QVcGfmF zHP=4=*!Js}2Bts#)9lB8^cd66V=)m-UB5JV{nA0$_1Yiay?XTQ^o5Is{>33rTl>~+ zo|$u>UOjpOwhoF)P)LZ1(i6v$r{42|j6>b2;s_|)(EZREKXOn`;iPmknZ8TUMw=-MlFH&&NyQB6!CbTzN7)}a8M=dLY)Ej5c06v5cN zd|}CN`>ORnuT7)lidL4yVoZuhwKB2}z@Rw{K_M6w1DZvS1>%ydS-RsmDyo{LOH}0M zwVDkSm{XEWM7*qKH35i;5izl!MoW7YVh|6Mo2~U~G*OS`L|m!wimO5@DhNFy;Hfq$ z#>D64s2o#km6W2&(Ru|3>J>SV#340R52^@j0F-r!l;)g=c6={vUkU$85~_amwS&Ln zSbnr|_~FA3k8~Z`+`o0owjgI)lCdor**&`N{TENZn5|iVq585tYisXsnRG14IaXvG zD<-WKN4MqN)fsp7=&r1LRnFazaW`b$51o5D*VLM6YQ5Z*ZF(Ztv?J5BBipodj-kqT zQ8y69?V{#DF$&y@Y%2`tJovfJ0~??&T;u?eHcAQkzk)~Vqx+aXwvS8G61{gFydafg z)B;8j3Kwu~t|hI+0v`zHZYxEU7P%7E-Rpd@G;5-?gywV+xkVmXRNh<{CXl+Vq%G>* zwQ!~u)2Vi%6-ccxw*lA6P-qWIF-5n^b!b$zW-)-c4U_G?c8yaKk&ueZ;TI+h#)Q0B z2RVVG0Yi-m#Z-;e!RrddqX`kqf)sjQR#eT_x-TH>DDsx!g|HaN0iIZvpbZ28eHupy zrD#@@Oq$hX6Ly2VLY|r+844zl)A-vQEnW$u_Yu)SsQS?j+R9Z&$u3!-wWP|t4$V0)!X=-9J#L9{p8SWyGar3?b^LuJ zROa6tgAjD}=-Bl58#9Bi%^V-k51p@{9X~aF@srupFJ1j`aOT+1%z+ceve`qQe-LDXzK;RIB z&{#_tJV&fXdCN72(B|tD9^1KR$F5&!{dQSg8CCs($npzj&%*>BKUBw&IzI@@J+Tu7Tdywv6n&;#f9Sv1H=DN3#{aiE`hx zt70fP7@SyCGq(8LFV8=F_Svj!YtGf4akXb%+Xt8_chzv);kJ>^(dw~9C+p7DoUc1u zmv#Gc?zW7(E$e=AfIVoNaxWfPdt$?z8?x>P2H4kanD~k!b)b((AlIpL1v_qP%&&jY#(Vn;u_#4o%amw8Qha|F3mWX zjwDAzxs}bCmCa{*vMaY-c4wX22IwiiGRH5=@XJQIG5VxE%dgJyjTyf2%(^VUal*Rs zhh{B$#1T;VV`eS+z&;v0417{8hU^|btk7ZvISca>OwFX(k}>aYbTAInfr$sf4@2eU z^l*=tkO9Lc5cNrBXc&PrjKHb!y})V)i#>QV5#0inLatki+}H;{Z?{5WvyDW&9V)Pb z%bjyA&A8yvv#tjQm`V4d5oV;~P}@jg^#0KmBf$X-*5MZpr3Oe-WVDDteBWITGl-&?NV z%#-$X#AEU;C$`+Ogt+t4jWo6^9*=lk_(>Qhx9Lv8_2D*{BqN5w2|abstaDOZed|YZ zT~O?WPm{U?4n;$1=mlA+3xpCu$m8lnS(J2ljfrp798W|Oajfc&U;k>*vo-zpn3%it zEShzb@ur*c4wxWuMENyTkWEq4zi5_Xzbi-7if@qfOXT_zIle@;8!SszeZ{s+R@~d) ze#4H|ZJyY*`#Z!@@OiMnMpgc&s*-Abb#)op=AP0P}ipm>ihKSxGrD_8vBevQ=bX?dd?6q2Q7V;pta8$wDs9AqRGoxW2$0ObF15sU<28w2NX`R7_7w zyx1E)#fKE@8D13nW2{ViezI4g=rjN?{{Vm~#G^h8+@m-Q+@d&|V>lhBe}yrj`+uB` zGn~`+FDCnSH@Y+F&FYX^_gN;mqT&%@cyVM$gyJr#|8Ky zpU6ks;Os3uM>?IW7?`*@_rlG0U%ma?mz~^Umy#pSyE*)(IT{?8|G3PfvvygUZL6pG_GVgb@hoPr+F zPPN2o=oQwYI0HShTAR~BFK@~S#r4q3t0zLTt$-zi@oVKmo+{uO3u7^`qAr(7p+OB4 zy~O)PUKUGXRxBfxgH+|DBKwGhyrN?V1X*$!6h`6${o*<@QAw&bmPrPfAZi8WUb|Ng zM!X^)2}|A+VVR8uS#}86CAx#YknbcPB*KznlFSFalIZusX;jts28+7yiNq8m%L*Yu zX4#QVUo{R-VUiq6Isnxe${A6``UEp>%v4rGW69L+Ofc7sHJR!L(r(OD)x)qovuP{o zZn)vudu2=7b11>QU|t3Ss3<$qDavjabcpN*i?SlqTH$bk*P~RN5_NHkqc*}+58~(^ zP^2zJ?+*xYU6F`ZF+@YBLgAqhI5hesFS}@Q0~yw-79>6dutc^69*TsDVrI3~tPmM_ zXe}NTr#F+eq#8q+vdXEAlN+b^q|53P`s>xTi8Co*GB$5)nl(WsZUU^eZ(0pEoI-N( zo`ZmSe<`>vxGx$k>gx(t9Fvoe(s5cm02&UVh4e=tL;-$Lhga}VwShu4#OXnzK=+lx zUKjzj%KtYtJl-Ii3Z&w6K|-*<8Yl}hT6^mIus@E$Ne7FTw1N?*3ac@K;Kp>Ra_s0O zMj;D9UR)2js@*7tNH`MU`(=d=@FCTn6g}7)AL3k$CT2>AC^U#)!KWp{2@rE$&K4Hp zR(Oadeq`j~wFJE~zXeuGXP_EGH%c56LQ?q2sRW&|mQEZ^9!-hKt^~ebVo$YBZ=P;> zxqkBKOvU2nriIN-S1Xz_R@+2pvUB|LyEL+weibEM(R9;Nk~sa7U8&wFc9Oknsn3*e z$QVr%#-uS-K5om{%Qf{`ElA>G&LV`DWzgM!q#s9EH5X0r9Cd0paHtxXUsNl=&lhlr zRFb|xbRNoC$WNXD_$6wJ{;^I6bYjHp^0OL_R+IQ!$T~*BRSp*2;R;rK#WcZvuBBda zT|u}KIbTs_R%j$?66GIwK4#Fef?t4unzaD0w1wS438 zv9|V@Ia}QN&xk8|l^!I1@V-)z$(POnw)GN2EMPa}5EwlLRxxV*2Wu?f4&fm*Oh1K& zFr4MvZ_z{8Rn`UG2y!2V8Ia|O!jKF_LBd;bZ^;Q;K2yu_vQG%qHrEml<5aU2CqL}x zd5-65M->C4KpZbhij5>jVNn=SGwl&WJ^>Ua38~FrCKVJUNeG>Eo>-M6){fpGetg8T zYReG{cCfC_4*jbIq6hr*#C===;HNn6KHzCW_yz8x0o=!dm6!yqUeh70m_@69qysye9^0|2 zv9WQVQwaG3QI7A4o@fgPeL_eAI(EPucK55Pa9DlpNNQARYwvmrriq3QyZ!mhm}rN) zfJGz}3QHlOzduh0Qd}OAchMaQ2LyjiA}%cyb{B@>`NRYSICKq+i(8>`=@dhdmn1l* zdNT!|Qt%m)D26zRIAapq#6@Of zC-5!UO5&jU;x_sMQXhL|>P+&7^R>IL+IQcmdU$&1wP#*_W_ItT6X}}0>8j?GVVTlB zP{J^O$(*9*BzCqq4VxVYt2<_ z%{_@E$!A;2TB`8}6}Y8}`Cy$MV3$oxjaB9l8xglb1=*{n0i0r1H9#U9SV^4~J;(dQ zV047nT3St9h@yvgC=9zw)SN+dk}-qG`*{JLs?7w~L#ju}xH$v{EV0C9#XV$nFR6&8 zDGV7@81fcVXs~o~AB;*Q+<;(y0RMc^LzG$ymu-D8`*X(tKhT+zu5D7^TK-feDf2DBMX&1newWseUtm9e|U*Xm+wflWy;r29h^Kk zv-euLJ5yOT6`Tytw9VGddwbHAy^E#2^QFC+(#jmg*w){u^~}nbS}q-)9iFRrZ_^(( zrEA+$47@_NxDxFt&w_RHUl@zYnX5wf@`-`uKq@+Ya{iH)bjjgG)0M{; zOOG#<9=}mq`M)_GEV!4eN2XSMPdW?oI$P7R2mcbcZ(&|<(QcHG{;e6XX%0G)qt zcC~NQ|3}?^fIqCk?TyTbP91@_ZfxIXr3f;$yHWcsi1jt#*hKdB;ME4lYj(ciurM z9D+9)apLeqhObRgNwuaC$Co|&i7*#)Q5qgYI{yk^m^9-&0&7LpE0J4+k*NkIl1Q?) zZRMTBcK|^;3YF@|{=396U)?nK?ToWyzUIhX8dr4UWrXcV@m~=(9mP4dRKDon zrb`Yj8V@ZP59M@Z{)xzp}Uk`XUsX!p2IlP)Sao@I=B6@#5yu+@rMHe_%200gRc5gqlfqoshCKtOEi$`FsVpD5y_iZ3o`0$zsv!}xL@P! zH^rv_kK|<%BwQ#*V~nq(?kuYP6j}a^w*47xxrKHvpq+o!cVpawKSye#p8Nj*0GF>| literal 0 HcmV?d00001 diff --git a/src/security/audit.py b/src/security/audit.py new file mode 100644 index 00000000..0e6c8926 --- /dev/null +++ b/src/security/audit.py @@ -0,0 +1,76 @@ +"""安全审计：操作日志记录、查询与不可篡改链式校验。""" + +from __future__ import annotations + +import hashlib +import json +import threading +import time +import uuid +from dataclasses import dataclass, field, asdict +from typing import Optional + + +@dataclass +class AuditEntry: + action: str # 操作类型，如 user.create / billing.refund + actor: str # 操作人 + target: str = "" # 操作对象 + detail: str = "" # 附加说明 + ip: str = "" + entry_id: str = field(default_factory=lambda: uuid.uuid4().hex) + ts: float = field(default_factory=time.time) + prev_hash: str = "" + entry_hash: str = "" + + def canonical(self) -> str: + d = asdict(self) + d.pop("entry_hash", None) + return json.dumps(d, sort_keys=True, ensure_ascii=False) + + +class AuditLogger: + """线程安全的内存审计日志（哈希链防篡改），生产可替换为 DB 存储。""" + + def __init__(self) -> None: + self._entries: list[AuditEntry] = [] + self._lock = threading.Lock() + + def log(self, action: str, actor: str, target: str = "", detail: str = "", ip: str = "") -> AuditEntry: + with self._lock: + prev = self._entries[-1].entry_hash if self._entries else "GENESIS" + entry = AuditEntry(action=action, actor=actor, target=target, detail=detail, ip=ip, + prev_hash=prev) + entry.entry_hash = hashlib.sha256(entry.canonical().encode("utf-8")).hexdigest() + self._entries.append(entry) + return entry + + def query(self, *, actor: Optional[str] = None, action: Optional[str] = None, + since: Optional[float] = None, until: Optional[float] = None) -> list[AuditEntry]: + with self._lock: + result = list(self._entries) + if actor is not None: + result = [e for e in result if e.actor == actor] + if action is not None: + result = [e for e in result if e.action == action] + if since is not None: + result = [e for e in result if e.ts >= since] + if until is not None: + result = [e for e in result if e.ts <= until] + return result + + def verify_chain(self) -> bool: + """校验哈希链完整性；任何条目被篡改都会返回 False""" + with self._lock: + entries = list(self._entries) + prev = "GENESIS" + for e in entries: + if e.prev_hash != prev: + return False + if hashlib.sha256(e.canonical().encode("utf-8")).hexdigest() != e.entry_hash: + return False + prev = e.entry_hash + return True + + def __len__(self) -> int: + return len(self._entries) diff --git a/src/security/auth.py b/src/security/auth.py new file mode 100644 index 00000000..90d19c7d --- /dev/null +++ b/src/security/auth.py @@ -0,0 +1,188 @@ +"""RBAC 角色权限 + JWT 认证（HS256，无第三方依赖）。 + +角色枚举：ADMIN / OPERATOR / VIEWER / DEVICE +权限模型：角色 -> 权限集合；@require_role / @require_permission 装饰器 +供 FastAPI 路由或普通函数使用。 +""" + +from __future__ import annotations + +import base64 +import functools +import hashlib +import hmac +import json +import time +import uuid +from dataclasses import dataclass, field +from enum import Enum +from typing import Callable, Iterable, Optional + + +class Role(str, Enum): + ADMIN = "admin" # 系统管理员：全部权限 + OPERATOR = "operator" # 业务操作员：业务读写 + VIEWER = "viewer" # 只读用户：查询 + DEVICE = "device" # 设备接入：仅数据上报 + + +class Permission(str, Enum): + USER_MANAGE = "user:manage" + DEVICE_READ = "device:read" + DEVICE_WRITE = "device:write" + DATA_REPORT = "data:report" + BILLING_READ = "billing:read" + BILLING_WRITE = "billing:write" + PATROL_READ = "patrol:read" + PATROL_WRITE = "patrol:write" + SYSTEM_SETTINGS = "system:settings" + AUDIT_READ = "audit:read" + + +ROLE_PERMISSIONS: dict[Role, frozenset[Permission]] = { + Role.ADMIN: frozenset(Permission), + Role.OPERATOR: frozenset({ + Permission.DEVICE_READ, Permission.DEVICE_WRITE, + Permission.BILLING_READ, Permission.BILLING_WRITE, + Permission.PATROL_READ, Permission.PATROL_WRITE, + Permission.DATA_REPORT, + }), + Role.VIEWER: frozenset({ + Permission.DEVICE_READ, Permission.BILLING_READ, Permission.PATROL_READ, + }), + Role.DEVICE: frozenset({Permission.DATA_REPORT}), +} + + +def _b64url_encode(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _b64url_decode(data: str) -> bytes: + pad = "=" * (-len(data) % 4) + return base64.urlsafe_b64decode(data + pad) + + +class AuthError(Exception): + """认证/授权失败""" + + +@dataclass +class UserToken: + user_id: str + username: str + role: Role + permissions: frozenset[Permission] = field(default_factory=frozenset) + + def has_permission(self, perm: Permission) -> bool: + return self.role == Role.ADMIN or perm in self.permissions + + +@dataclass +class TokenPair: + access_token: str + refresh_token: str + token_type: str = "bearer" + expires_in: int = 3600 + + +def _sign(payload: dict, secret: str) -> str: + header = {"alg": "HS256", "typ": "JWT"} + segments = [ + _b64url_encode(json.dumps(header, separators=(",", ":")).encode()), + _b64url_encode(json.dumps(payload, separators=(",", ":")).encode()), + ] + signing_input = ".".join(segments).encode() + sig = hmac.new(secret.encode(), signing_input, hashlib.sha256).digest() + segments.append(_b64url_encode(sig)) + return ".".join(segments) + + +def _decode(token: str, secret: str, verify_exp: bool = True) -> dict: + try: + head_b64, body_b64, sig_b64 = token.split(".") + except ValueError as exc: + raise AuthError("token 格式非法") from exc + signing_input = f"{head_b64}.{body_b64}".encode() + expected = hmac.new(secret.encode(), signing_input, hashlib.sha256).digest() + try: + actual = _b64url_decode(sig_b64) + except Exception as exc: + raise AuthError("token 签名非法") from exc + if not hmac.compare_digest(expected, actual): + raise AuthError("token 签名校验失败") + payload = json.loads(_b64url_decode(body_b64)) + if verify_exp and "exp" in payload and time.time() > payload["exp"]: + raise AuthError("token 已过期") + return payload + + +def create_tokens(user_id: str, username: str, role: Role, secret: str, + access_ttl: int = 3600, refresh_ttl: int = 86400 * 7) -> TokenPair: + """生成 access/refresh token 对""" + now = int(time.time()) + jti = uuid.uuid4().hex + access = _sign({ + "sub": user_id, "username": username, "role": role.value, + "iat": now, "exp": now + access_ttl, "jti": jti, "type": "access", + }, secret) + refresh = _sign({ + "sub": user_id, "username": username, "role": role.value, + "iat": now, "exp": now + refresh_ttl, + "jti": uuid.uuid4().hex, "type": "refresh", + }, secret) + return TokenPair(access_token=access, refresh_token=refresh, expires_in=access_ttl) + + +def verify_token(token: str, secret: str, expected_type: str = "access") -> UserToken: + """校验 token 并还原用户上下文""" + payload = _decode(token, secret) + if payload.get("type") != expected_type: + raise AuthError(f"token 类型错误，期望 {expected_type}") + role = Role(payload["role"]) + return UserToken( + user_id=payload["sub"], + username=payload.get("username", ""), + role=role, + permissions=ROLE_PERMISSIONS.get(role, frozenset()), + ) + + +def _extract_user(args, kwargs) -> Optional[UserToken]: + if "current_user" in kwargs: + return kwargs["current_user"] + for a in args: + if isinstance(a, UserToken): + return a + return None + + +def require_role(*roles: Role) -> Callable: + """装饰器：要求当前用户角色在允许列表内""" + def deco(func: Callable) -> Callable: + @functools.wraps(func) + def wrapper(*args, **kwargs): + user = _extract_user(args, kwargs) + if user is None: + raise AuthError("缺少当前用户上下文（current_user）") + if user.role not in roles: + raise AuthError(f"角色 {user.role.value} 无权执行该操作") + return func(*args, **kwargs) + return wrapper + return deco + + +def require_permission(*perms: Permission) -> Callable: + """装饰器：要求当前用户具备全部指定权限""" + def deco(func: Callable) -> Callable: + @functools.wraps(func) + def wrapper(*args, **kwargs): + user = _extract_user(args, kwargs) + if user is None: + raise AuthError("缺少当前用户上下文（current_user）") + missing = [p for p in perms if not user.has_permission(p)] + if missing: + raise AuthError("缺少权限: " + ",".join(p.value for p in missing)) + return func(*args, **kwargs) + return wrapper + return deco diff --git a/src/security/encryption.py b/src/security/encryption.py new file mode 100644 index 00000000..d1ddaf72 --- /dev/null +++ b/src/security/encryption.py @@ -0,0 +1,62 @@ +"""数据加密与口令散列工具。 + +- DataEncryptor：AES-256-GCM（依赖 cryptography），密钥由主密钥经 HKDF 派生 +- hash_password / verify_password：PBKDF2-HMAC-SHA256（stdlib） +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import os + +try: + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + _HAS_CRYPTO = True +except ImportError: # pragma: no cover + _HAS_CRYPTO = False + + +def hash_password(password: str, *, iterations: int = 120_000, salt: bytes | None = None) -> str: + """PBKDF2-HMAC-SHA256 口令散列，输出 `pbkdf2$iterations$salt_b64$hash_b64`""" + salt = salt or os.urandom(16) + dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, iterations) + return "pbkdf2${}${}${}".format( + iterations, base64.b64encode(salt).decode(), base64.b64encode(dk).decode()) + + +def verify_password(password: str, stored: str) -> bool: + try: + scheme, iters, salt_b64, hash_b64 = stored.split("$") + if scheme != "pbkdf2": + return False + dk = hashlib.pbkdf2_hmac("sha256", password.encode(), + base64.b64decode(salt_b64), int(iters)) + return hmac.compare_digest(dk, base64.b64decode(hash_b64)) + except Exception: + return False + + +class DataEncryptor: + """AES-256-GCM 字段级加密器（用于手机号/身份证等敏感字段落库加密）。""" + + def __init__(self, master_key: bytes, *, info: bytes = b"wms-field-encryption"): + if not _HAS_CRYPTO: + raise RuntimeError("需要安装 cryptography 库以使用 AES-256-GCM 加密") + if len(master_key) < 16: + raise ValueError("master_key 长度至少 16 字节") + # HKDF-SHA256 派生 32 字节数据密钥 + prk = hmac.new(b"wms-hkdf-salt", master_key, hashlib.sha256).digest() + self._key = hmac.new(prk, info + b"\x01", hashlib.sha256).digest() + self._aes = AESGCM(self._key) + + def encrypt(self, plaintext: str) -> str: + nonce = os.urandom(12) + ct = self._aes.encrypt(nonce, plaintext.encode("utf-8"), None) + return base64.b64encode(nonce + ct).decode("ascii") + + def decrypt(self, token: str) -> str: + raw = base64.b64decode(token) + nonce, ct = raw[:12], raw[12:] + return self._aes.decrypt(nonce, ct, None).decode("utf-8") diff --git a/src/security/middleware.py b/src/security/middleware.py new file mode 100644 index 00000000..044a96f2 --- /dev/null +++ b/src/security/middleware.py @@ -0,0 +1,91 @@ +"""FastAPI 安全中间件：JWT 认证、CORS、限流、安全响应头。""" + +from __future__ import annotations + +import time +from collections import defaultdict, deque +from typing import Callable, Optional + +from .auth import AuthError, UserToken, verify_token + + +class SecurityMiddleware: + """ASGI 中间件（不依赖 starlette 内部 API，可挂载到任意 ASGI app）。 + + 功能： + 1. 受保护路径的 Bearer token 认证，解析后放入 scope["current_user"] + 2. 简单滑动窗口限流（按客户端 IP） + 3. 统一安全响应头（HSTS/X-Frame-Options/X-Content-Type-Options/CSP） + """ + + def __init__(self, app, *, secret: str, + protected_prefixes: tuple[str, ...] = ("/api/",), + exempt_paths: tuple[str, ...] = ("/api/auth/login", "/health"), + rate_limit: int = 120, rate_window: int = 60) -> None: + self.app = app + self.secret = secret + self.protected_prefixes = protected_prefixes + self.exempt_paths = exempt_paths + self.rate_limit = rate_limit + self.rate_window = rate_window + self._hits: dict[str, deque[float]] = defaultdict(deque) + + def _client_ip(self, scope) -> str: + client = scope.get("client") + return client[0] if client else "unknown" + + def _allow_rate(self, ip: str) -> bool: + now = time.time() + q = self._hits[ip] + while q and now - q[0] > self.rate_window: + q.popleft() + if len(q) >= self.rate_limit: + return False + q.append(now) + return True + + async def __call__(self, scope, receive, send) -> None: + if scope["type"] != "http": + await self.app(scope, receive, send) + return + path = scope.get("path", "") + ip = self._client_ip(scope) + + if not self._allow_rate(ip): + await self._respond(send, 429, b'{"detail":"rate limit exceeded"}') + return + + needs_auth = path.startswith(self.protected_prefixes) and path not in self.exempt_paths + if needs_auth: + headers = {k.decode(): v.decode() for k, v in scope.get("headers", [])} + auth = headers.get("authorization", "") + if not auth.startswith("Bearer "): + await self._respond(send, 401, b'{"detail":"missing bearer token"}') + return + try: + user: UserToken = verify_token(auth[7:], self.secret) + except AuthError as exc: + await self._respond(send, 401, f'{{"detail":"{exc}"}}'.encode()) + return + scope["current_user"] = user + + async def send_with_security_headers(message): + if message["type"] == "http.response.start": + headers = list(message.get("headers", [])) + for k, v in [ + (b"strict-transport-security", b"max-age=31536000; includeSubDomains"), + (b"x-frame-options", b"DENY"), + (b"x-content-type-options", b"nosniff"), + (b"referrer-policy", b"no-referrer"), + ]: + headers.append((k, v)) + message["headers"] = headers + await send(message) + + await self.app(scope, receive, send_with_security_headers) + + @staticmethod + async def _respond(send, status: int, body: bytes) -> None: + await send({"type": "http.response.start", "status": status, + "headers": [(b"content-type", b"application/json")]}) + await send({"type": "http.response.body", "body": body}) diff --git a/tests/security/test_security.py b/tests/security/test_security.py new file mode 100644 index 00000000..f9cbcd18 --- /dev/null +++ b/tests/security/test_security.py @@ -0,0 +1,186 @@ +"""src/security 单元测试：JWT/RBAC/加密/审计/中间件（18 个用例）""" + +import time +import unittest + +from src.security.auth import ( + AuthError, Permission, Role, UserToken, + create_tokens, verify_token, require_role, require_permission, +) +from src.security.encryption import DataEncryptor, hash_password, verify_password +from src.security.audit import AuditLogger +from src.security.middleware import SecurityMiddleware + +SECRET = "unit-test-secret-key-32bytes-long!" + + +class TestJwtAuth(unittest.TestCase): + def test_create_and_verify_access_token(self): + pair = create_tokens("u1", "zhangsan", Role.OPERATOR, SECRET) + user = verify_token(pair.access_token, SECRET) + self.assertEqual(user.user_id, "u1") + self.assertEqual(user.role, Role.OPERATOR) + self.assertEqual(pair.token_type, "bearer") + + def test_refresh_token_type(self): + pair = create_tokens("u1", "zhangsan", Role.ADMIN, SECRET) + user = verify_token(pair.refresh_token, SECRET, expected_type="refresh") + self.assertEqual(user.user_id, "u1") + with self.assertRaises(AuthError): + verify_token(pair.refresh_token, SECRET) # 类型不匹配 + + def test_expired_token_rejected(self): + pair = create_tokens("u1", "zhangsan", Role.VIEWER, SECRET, access_ttl=-1) + with self.assertRaises(AuthError): + verify_token(pair.access_token, SECRET) + + def test_tampered_token_rejected(self): + pair = create_tokens("u1", "zhangsan", Role.VIEWER, SECRET) + bad = pair.access_token[:-2] + "xx" + with self.assertRaises(AuthError): + verify_token(bad, SECRET) + + def test_wrong_secret_rejected(self): + pair = create_tokens("u1", "zhangsan", Role.VIEWER, SECRET) + with self.assertRaises(AuthError): + verify_token(pair.access_token, "another-secret") + + +class TestRBAC(unittest.TestCase): + def _user(self, role): + return UserToken("u", "n", role, { + Role.ADMIN: frozenset(Permission), + Role.OPERATOR: frozenset({Permission.BILLING_READ, Permission.BILLING_WRITE}), + Role.VIEWER: frozenset({Permission.BILLING_READ}), + Role.DEVICE: frozenset({Permission.DATA_REPORT}), + }[role]) + + def test_admin_has_all_permissions(self): + self.assertTrue(self._user(Role.ADMIN).has_permission(Permission.USER_MANAGE)) + + def test_viewer_cannot_write(self): + self.assertFalse(self._user(Role.VIEWER).has_permission(Permission.BILLING_WRITE)) + + def test_require_role_allows(self): + @require_role(Role.ADMIN, Role.OPERATOR) + def op(current_user=None): + return "ok" + self.assertEqual(op(current_user=self._user(Role.OPERATOR)), "ok") + + def test_require_role_denies(self): + @require_role(Role.ADMIN) + def op(current_user=None): + return "ok" + with self.assertRaises(AuthError): + op(current_user=self._user(Role.VIEWER)) + + def test_require_permission_denies_missing(self): + @require_permission(Permission.BILLING_WRITE) + def op(current_user=None): + return "ok" + with self.assertRaises(AuthError): + op(current_user=self._user(Role.VIEWER)) + self.assertEqual(op(current_user=self._user(Role.OPERATOR)), "ok") + + +class TestEncryption(unittest.TestCase): + def test_password_hash_roundtrip(self): + stored = hash_password("S3cret!") + self.assertTrue(verify_password("S3cret!", stored)) + self.assertFalse(verify_password("wrong", stored)) + + def test_password_hash_unique_salt(self): + self.assertNotEqual(hash_password("same"), hash_password("same")) + + def test_aesgcm_roundtrip(self): + enc = DataEncryptor(b"master-key-for-testing-32b") + token = enc.encrypt("13800138000") + self.assertEqual(enc.decrypt(token), "13800138000") + + def test_aesgcm_tamper_detected(self): + enc = DataEncryptor(b"master-key-for-testing-32b") + token = enc.encrypt("sensitive") + import base64 + raw = bytearray(base64.b64decode(token)); raw[-1] ^= 1 + with self.assertRaises(Exception): + enc.decrypt(base64.b64encode(bytes(raw)).decode()) + + def test_short_master_key_rejected(self): + with self.assertRaises(ValueError): + DataEncryptor(b"short") + + +class TestAudit(unittest.TestCase): + def test_log_and_query(self): + log = AuditLogger() + log.log("user.create", "admin", target="u100") + log.log("billing.refund", "operator", target="r9", ip="10.0.0.1") + self.assertEqual(len(log), 2) + self.assertEqual(len(log.query(actor="admin")), 1) + self.assertEqual(log.query(action="billing.refund")[0].ip, "10.0.0.1") + + def test_hash_chain_integrity(self): + log = AuditLogger() + for i in range(5): + log.log(f"op.{i}", "tester") + self.assertTrue(log.verify_chain()) + + def test_hash_chain_tamper_detected(self): + log = AuditLogger() + log.log("a", "x"); log.log("b", "x") + log._entries[0].detail = "tampered" + self.assertFalse(log.verify_chain()) + + +class TestMiddleware(unittest.IsolatedAsyncioTestCase): + def _make_scope(self, path="/api/devices", token=None): + headers = [] + if token: + headers.append((b"authorization", f"Bearer {token}".encode())) + return {"type": "http", "path": path, "headers": headers, + "client": ("127.0.0.1", 12345)} + + @staticmethod + def _send_collector(sent): + async def _send(message): + sent.append(message) + return _send + + async def test_missing_token_401(self): + async def app(scope, receive, send): + raise AssertionError("不应进入业务") + mw = SecurityMiddleware(app, secret=SECRET) + sent = [] + await mw(self._make_scope(), None, self._send_collector(sent)) + self.assertEqual(sent[0]["status"], 401) + + async def test_valid_token_passes_and_headers(self): + async def app(scope, receive, send): + assert "current_user" in scope + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"ok"}) + mw = SecurityMiddleware(app, secret=SECRET) + pair = create_tokens("u1", "op", Role.OPERATOR, SECRET) + sent = [] + await mw(self._make_scope(token=pair.access_token), None, self._send_collector(sent)) + self.assertEqual(sent[0]["status"], 200) + hdr_keys = {k for k, _ in sent[0]["headers"]} + self.assertIn(b"strict-transport-security", hdr_keys) + self.assertIn(b"x-frame-options", hdr_keys) + + async def test_rate_limit(self): + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"ok"}) + mw = SecurityMiddleware(app, secret=SECRET, rate_limit=3, rate_window=60) + pair = create_tokens("u1", "op", Role.OPERATOR, SECRET) + statuses = [] + for _ in range(5): + sent = [] + await mw(self._make_scope(token=pair.access_token), None, self._send_collector(sent)) + statuses.append(sent[0]["status"]) + self.assertEqual(statuses.count(429), 2) + + +if __name__ == "__main__": + unittest.main() -- 2.54.0