(self, pkt=None)
| 1124 | |
| 1125 | @ATMT.state() |
| 1126 | def AUTH(self, pkt=None): |
| 1127 | if pkt is None: |
| 1128 | return |
| 1129 | if HTTPRequest in pkt: |
| 1130 | self.vprint(pkt.summary()) |
| 1131 | if pkt.Method == b"CONNECT": |
| 1132 | # HTTP tunnel (proxy) |
| 1133 | proxy = True |
| 1134 | else: |
| 1135 | # HTTP non-tunnel |
| 1136 | proxy = False |
| 1137 | # Get authorization |
| 1138 | if proxy: |
| 1139 | authorization = pkt.Proxy_Authorization |
| 1140 | else: |
| 1141 | authorization = pkt.Authorization |
| 1142 | if not authorization: |
| 1143 | # Initial ask. |
| 1144 | data = self.authmethod |
| 1145 | if self.basic: |
| 1146 | data += " realm='%s'" % self.BASIC_REALM |
| 1147 | self._ask_authorization(proxy, data) |
| 1148 | return |
| 1149 | # Parse authorization |
| 1150 | method, data = authorization.split(b" ", 1) |
| 1151 | if plain_str(method) != self.authmethod: |
| 1152 | self.debug(3, "Bad auth method.") |
| 1153 | raise self.AUTH_ERROR(proxy) |
| 1154 | try: |
| 1155 | data = base64.b64decode(data) |
| 1156 | except Exception: |
| 1157 | self.debug(3, "Couldn't unpack base64 of auth.") |
| 1158 | raise self.AUTH_ERROR(proxy) |
| 1159 | # Now process the authorization |
| 1160 | if not self.basic: |
| 1161 | try: |
| 1162 | ssp_blob = GSSAPI_BLOB(data) |
| 1163 | except Exception: |
| 1164 | self.sspcontext = None |
| 1165 | self._ask_authorization(proxy, self.authmethod) |
| 1166 | self.debug(3, "Couldn't unpack GSSAPI_BLOB of auth.") |
| 1167 | raise self.AUTH_ERROR(proxy) |
| 1168 | # And call the SSP |
| 1169 | self.sspcontext, tok, status = self.ssp.GSS_Accept_sec_context( |
| 1170 | self.sspcontext, |
| 1171 | ssp_blob, |
| 1172 | req_flags=self.ssp_req_flags, |
| 1173 | chan_bindings=self.chan_bindings, |
| 1174 | ) |
| 1175 | else: |
| 1176 | # This is actually Basic authentication |
| 1177 | try: |
| 1178 | next( |
| 1179 | True |
| 1180 | for k, v in self.BASIC_IDENTITIES.items() |
| 1181 | if ("%s:%s" % (k, v)).encode() == data |
| 1182 | ) |
| 1183 | tok, status = None, GSS_S_COMPLETE |
no test coverage detected