Coverage for packages/azericard/src/integrify/azericard/schemas/request/auth.py: 100%

1import base64 

2import json 

3from functools import cache 

4from typing import ClassVar, Literal 

5 

6import rsa 

7from integrify.azericard import env 

8from integrify.azericard.schemas.common import ( 

9 AzeriCardMinimalDataSchema, 

10 AzeriCardMinimalWithAmountDataSchema, 

11) 

12from integrify.azericard.schemas.enums import ( 

13 AuthorizationMiscType, 

14 AuthorizationResponseType, 

15 AuthorizationType, 

16) 

17from integrify.schemas import PayloadBaseModel 

18from pydantic import ( 

19 AliasGenerator, 

20 ConfigDict, 

21 Field, 

22 computed_field, 

23 field_serializer, 

24) 

25from typing_extensions import TypedDict 

26 

27 

28@cache 

29def _load_private_key(key_file_path: str) -> rsa.PrivateKey: 

30 """RSA private key-i fayldan oxuyub cache-ləyən funksiya. 

31 

32 Əvvəllər hər sorğuda fayl oxunub parse olunurdu; indi eyni fayl yolu üçün bir dəfə 

33 oxunur. Nəticə eynidir (deterministik), lakin hər sorğuda disk I/O aradan qalxır. 

34 """ 

35 with open(key_file_path, 'rb') as key_file: 

36 return rsa.PrivateKey.load_pkcs1(key_file.read()) 

37 

38 

39class BaseRequestSchema(PayloadBaseModel): 

40 SIGNATURE_FIELDS: ClassVar[list[str]] 

41 """P_SIGN hesablanılması üçün lazım olan field adları""" 

42 

43 model_config = ConfigDict(alias_generator=AliasGenerator(serialization_alias=str.upper)) 

44 

45 @computed_field 

46 def p_sign(self) -> str | None: 

47 """P_SIGN generasiyası""" 

48 if not self.SIGNATURE_FIELDS: 

49 return None # pragma: no cover 

50 

51 private_key = _load_private_key(env.AZERICARD_KEY_FILE_PATH) 

52 

53 mac_source = self.generate_mac_source().encode('utf-8') 

54 return rsa.sign(mac_source, private_key, 'SHA-256').hex() 

55 

56 def generate_mac_source(self): 

57 """P_SIGN üçün MAC source-un yaradılması""" 

58 source = '' 

59 for field in self.SIGNATURE_FIELDS: 

60 val = getattr(self, field) 

61 

62 if val: 

63 source += str(len(str(val))) + str(val) 

64 else: 

65 # So far, no case can reach this state 

66 source += '-' # pragma: no cover 

67 

68 return source 

69 

70 

71class MobilePhone(TypedDict): 

72 cc: str 

73 subscriber: str 

74 

75 

76class MInfo(TypedDict): 

77 browserScreenHeight: str 

78 browserScreenWidth: str 

79 browserTZ: str 

80 mobilePhone: MobilePhone 

81 

82 

83class AuthRequestSchema(BaseRequestSchema, AzeriCardMinimalWithAmountDataSchema): 

84 SIGNATURE_FIELDS: ClassVar[list[str]] = [ 

85 'amount', 

86 'currency', 

87 'terminal', 

88 'trtype', 

89 'timestamp', 

90 'nonce', 

91 'merch_url', 

92 ] 

93 

94 # Next three fields can be set either through 

95 # functions or environment, but they MUST be set 

96 desc: str = Field(default=env.AZERICARD_MERCHANT_NAME, min_length=1, max_length=50) 

97 merch_name: str = Field(default=env.AZERICARD_MERCHANT_NAME, min_length=1, max_length=50) 

98 merch_url: str = Field(default=env.AZERICARD_MERCHANT_URL, min_length=1, max_length=250) 

99 

100 email: str | None = Field(default=env.AZERICARD_MERCHANT_EMAIL, max_length=80) 

101 country: str | None = Field(None, max_length=2) 

102 merch_gmt: str | None = Field(None, min_length=1, max_length=5) 

103 backref: str = Field(default=env.AZERICARD_CALLBACK_URL, min_length=1, max_length=250) 

104 lang: str = Field(default=env.AZERICARD_INTERFACE_LANG, min_length=2, max_length=2) 

105 name: str | None = Field(None, min_length=2, max_length=45) 

106 m_info: MInfo | None = None 

107 

108 @field_serializer('m_info') 

109 def serialize_minfo_to_b64(self, m_info: MInfo | None): 

110 """M_INFO dcit-ini base64 encodelaşdırılması""" 

111 if not m_info: 

112 return None 

113 

114 return base64.b64encode(json.dumps(m_info).encode()).decode() 

115 

116 @classmethod 

117 def get_input_fields(cls): 

118 return [ 

119 'amount', 

120 'currency', 

121 'order', 

122 'desc', 

123 'trtype', 

124 'merch_name', 

125 'merch_url', 

126 'terminal', 

127 'email', 

128 'country', 

129 'merch_gmt', 

130 'backref', 

131 'timestamp', 

132 'lang', 

133 'name', 

134 'm_info', 

135 ] 

136 

137 

138class AuthConfirmRequestSchema(BaseRequestSchema, AzeriCardMinimalWithAmountDataSchema): 

139 SIGNATURE_FIELDS: ClassVar[list[str]] = [ 

140 'order', 

141 'amount', 

142 'currency', 

143 'terminal', 

144 'trtype', 

145 'rrn', 

146 'int_ref', 

147 ] 

148 

149 rrn: str = Field(min_length=12, max_length=12) 

150 """Müştəri bankının axtarış istinad nömrəsi (ISO-8583 Sahə 37)""" 

151 

152 int_ref: str = Field(min_length=1, max_length=128) 

153 """Elektron ticarət şlüzünün daxili istinad nömrəsi""" 

154 

155 @classmethod 

156 def get_input_fields(cls): 

157 return [ 

158 'order', 

159 'amount', 

160 'currency', 

161 'rrn', 

162 'int_ref', 

163 'trtype', 

164 'terminal', 

165 'timestamp', 

166 ] 

167 

168 

169class AuthAndSaveCardRequestSchema(AuthRequestSchema): 

170 token_action: Literal['REGISTER'] = 'REGISTER' 

171 

172 

173class AuthWithSavedCardRequestSchema(AuthRequestSchema): 

174 token: str = Field(min_length=28, max_length=28) 

175 

176 @classmethod 

177 def get_input_fields(cls): 

178 return [ 

179 'amount', 

180 'currency', 

181 'order', 

182 'desc', 

183 'trtype', 

184 'token', 

185 'merch_name', 

186 'merch_url', 

187 'terminal', 

188 'email', 

189 'country', 

190 'merch_gmt', 

191 'backref', 

192 'timestamp', 

193 'lang', 

194 'name', 

195 'm_info', 

196 ] 

197 

198 

199class GetTransactionStatusRequestSchema(BaseRequestSchema, AzeriCardMinimalDataSchema): 

200 SIGNATURE_FIELDS: ClassVar[list[str]] = [ 

201 'order', 

202 'terminal', 

203 'trtype', 

204 'timestamp', 

205 'nonce', 

206 ] 

207 tran_trtype: AuthorizationType | AuthorizationResponseType = Field( 

208 min_length=1, 

209 max_length=2, 

210 ) 

211 trtype: Literal[AuthorizationMiscType.REQUEST_STATUS] = AuthorizationMiscType.REQUEST_STATUS 

212 

213 @classmethod 

214 def get_input_fields(cls): 

215 return ['tran_trtype', 'order', 'terminal', 'timestamp']