diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index aa51cee..f7adbae 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -11,7 +11,7 @@ jobs: build: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v9 + - uses: actions/checkout@v6 - name: Set up Python 3.12 uses: actions/setup-python@v6 with: diff --git a/aiowmi/connection.py b/aiowmi/connection.py index e057080..46c7d83 100644 --- a/aiowmi/connection.py +++ b/aiowmi/connection.py @@ -1,14 +1,14 @@ import asyncio from typing import Optional, Callable from Crypto.Cipher import ARC4 -from .exceptions import AccessDenied, NoNewActiveKey, BindNak +from .exceptions import AccessDenied, BindNak from .kerberos.ap_req import build_ap_req, wrap_gss_kerberos, get_active_key from .kerberos.cache import KerberosCache from .kerberos.krb5_pdu import get_neg_token, build_alter_context from .kerberos.tgs import get_tgs from .kerberos.tgt import get_tgt -from .kerberos.tools import gss_unwrap_kerberos -from .kerberos.tools import sign_func_kerberos, seal_func_kerberos +from .kerberos.wrappers import sign_func_kerberos, seal_func_kerberos +from .kerberos.wrappers import gss_unwrap_kerberos from .ntlm.auth_authenticate import NTLMAuthAuthenticate from .ntlm.auth_challange import NTLMAuthChallenge from .ntlm.auth_negotiate import NTLMAuthNegotiate @@ -206,51 +206,37 @@ async def _negotiate_kerberos(self): self._kerberos_cache.write(self._tgt, self._tgs, logger) - async def negotiate_kerberos(self, max_retry: int = 3) -> Protocol: + async def negotiate_kerberos(self) -> Protocol: if not self._domain: raise Exception('domain is required for Kerberos authentication') - attempt = 1 - has_keys = self._tgt is not None and self._tgt is not None - while True: - if self._protocol is None: - await self.connect(timeout=self._timeout) - assert self._protocol - proto = self._protocol - proto._auth_level = RPC_C_AUTHN_LEVEL_PKT_PRIVACY - proto._context_id = 0x0001357f - - if not has_keys: - logger.info('Start Kerberos negotion for TGT/TGS') - await self._negotiate_kerberos() - has_keys = True - else: - logger.info('Using Kerberos TGT/TGS from cache') - try: - await self._bind_kerberos(IID_IRemoteSCMActivator, proto) - iface = await self._if_binding(proto, - bind_func=self._bind_kerberos, - m_auth_level=proto._auth_level) - except (AccessDenied, NoNewActiveKey, BindNak) as e: - msg = str(e) or type(e).__name__ - if attempt <= max_retry: - logger.info(f'{msg} (attempt {attempt}/{max_retry})') - if attempt % 2 == 0: - has_keys = False # attemp to get new keys - self.close() - await asyncio.sleep(0.1 * attempt) - attempt += 1 - continue - raise - else: - break + has_keys = self._tgt is not None and self._tgs is not None + if self._protocol is None: + await self.connect(timeout=self._timeout) + assert self._protocol + proto = self._protocol + proto._auth_level = RPC_C_AUTHN_LEVEL_PKT_PRIVACY + proto._context_id = 79231 + if not has_keys: + logger.info('Start Kerberos negotiation for TGT/TGS') + await self._negotiate_kerberos() + has_keys = True + else: + logger.info('Using Kerberos TGT/TGS from cache') + + await self._bind_kerberos(IID_IRemoteSCMActivator, proto) + iface = await self._if_binding(proto, + bind_func=self._bind_kerberos, + m_auth_level=proto._auth_level) return iface async def _bind_kerberos(self, iid: bytes, proto: Protocol): - ticket, service_session_key = self._tgs + ticket, service_session_key, etype = self._tgs ap_req = build_ap_req(self._username, self._domain, - ticket, service_session_key) + ticket, + service_session_key, + etype) auth_value = wrap_gss_kerberos(ap_req) bind_pkg = proto._dcom.get_bind_kerberos_pkg( iid, @@ -260,13 +246,22 @@ async def _bind_kerberos(self, iid: bytes, proto: Protocol): ) rpc_bind_ack = await proto.get_dcom_response(bind_pkg) active_key, seq_number = get_active_key(rpc_bind_ack.auth.auth_value, - service_session_key) + service_session_key, + etype) if active_key is None: - raise NoNewActiveKey() + logger.warning("No new active key, use existing service key") + active_key = service_session_key + + if seq_number is None: + logger.warning( + "No new seq_number; " + f"proto._dcom._seq_num = {proto._dcom._seq_num}") + seq_number = proto._dcom.get_seq_num() + proto._auth_type = rpc_bind_ack.auth.auth_type proto._auth_level = rpc_bind_ack.auth.auth_level - neg_token = get_neg_token(service_session_key, seq_number) + neg_token = get_neg_token(service_session_key, seq_number, etype) alter_context_pkg = build_alter_context( iid, proto._dcom.get_new_call_id(), @@ -275,11 +270,10 @@ async def _bind_kerberos(self, iid: bytes, proto: Protocol): neg_token, ) _ = await proto.get_dcom_response(alter_context_pkg) - - if proto._auth_level >= RPC_C_AUTHN_LEVEL_PKT_INTEGRITY: - proto._client_sign = sign_func_kerberos(active_key) - proto._client_seal = seal_func_kerberos(active_key) - proto._server_seal = gss_unwrap_kerberos(active_key) + if active_key and proto._auth_level >= RPC_C_AUTHN_LEVEL_PKT_INTEGRITY: + proto._client_sign = sign_func_kerberos(active_key, etype) + proto._client_seal = seal_func_kerberos(active_key, etype) + proto._server_seal = gss_unwrap_kerberos(active_key, etype) async def negotiate_ntlm(self) -> Protocol: proto = self._protocol @@ -304,7 +298,7 @@ async def login_ntlm( request.set_pdu_data(ntlm_login_pkg) request_pkg = request.sign_data(proto) - + # print(request_pkg) rpc_response: RpcResponse = \ await proto.get_dcom_response( request_pkg, @@ -329,6 +323,7 @@ async def _if_binding(self, IID_IWbemLevel1Login) rci_pkg = remote_create_instance.get_data() + context_id = proto._context_id request = RpcRequest(op_num=4) request.set_pdu_data(rci_pkg) @@ -376,6 +371,6 @@ async def _if_binding(self, proto._auth_level = max( interface.scm_reply_info_data.authn_hint, m_auth_level) - + proto._context_id = context_id + 1 await bind_func(IID_IWbemLevel1Login, proto) return proto diff --git a/aiowmi/kerberos/ap_req.py b/aiowmi/kerberos/ap_req.py index 0a4bfa2..0cfaeec 100644 --- a/aiowmi/kerberos/ap_req.py +++ b/aiowmi/kerberos/ap_req.py @@ -1,35 +1,37 @@ -from typing import Optional +from typing import Optional, Tuple from datetime import datetime, timezone from .asn1 import asn1_len, asn1_tag, asn1_gt, asn1_int, asn1_ostr -from .tools import encrypt_kerberos_rc4, decrypt_kerberos_rc4 -from .const import OID_KERBEROS_V5, OID_MS_LEGACY_KRB, OID_SPNEGO +from .tools import encrypt_kerberos_rc4, decrypt_kerberos_rc4, peel_tag +from .tools import decrypt_kerberos_aes_cts, encrypt_kerberos_aes_cts +from .const import OID_SPNEGO, OID_IETF_KRB5, OID_MS_KRB5 +from ..exceptions import NoNewActiveKey -def wrap_gss_kerberos(ap_req_bytes): - # MechTypes [0] - mech_types = asn1_tag(0, b'\x30\x0b' + OID_MS_LEGACY_KRB) +def wrap_gss_kerberos(ap_req_bytes: bytes): + outer_oid = OID_IETF_KRB5 + inner_oid = OID_MS_KRB5 - # MechToken [2] -> GSS-API Token wrapper (0x60) - inner_token = OID_KERBEROS_V5 + b'\x01\x00' + ap_req_bytes - gss_token = b'\x60' + asn1_len(inner_token) + inner_token + mech_types_inner = b'\x30' + asn1_len(outer_oid) + outer_oid + mech_types = asn1_tag(0, mech_types_inner) - # OctetString wrapper (0x04) MechToken [2] - mech_token = asn1_tag(2, asn1_ostr(gss_token)) + inner_gss_content = inner_oid + b'\x01\x00' + ap_req_bytes + gss_wrapper = b'\x60' + asn1_len(inner_gss_content) + inner_gss_content - # NegTokenInit Sequence [30] -> Wrapper [0] - neg_token_body = mech_types + mech_token - neg_token_seq = b'\x30' + asn1_len(neg_token_body) + neg_token_body - neg_token_init = asn1_tag(0, neg_token_seq) + mech_token = asn1_tag(2, asn1_ostr(gss_wrapper)) - # Application 0 (0x60) wrapper - final_body = OID_SPNEGO + neg_token_init - return b'\x60' + asn1_len(final_body) + final_body + neg_body = mech_types + mech_token + neg_init = asn1_tag(0, b'\x30' + asn1_len(neg_body) + neg_body) + + final_payload = OID_SPNEGO + neg_init + + return b'\x60' + asn1_len(final_payload) + final_payload def build_ap_req(username: str, domain: str, ticket: bytes, - service_session_key: bytes) -> bytes: + service_session_key: bytes, + etype: int) -> bytes: now = datetime.now(timezone.utc) timestamp = now.strftime('%Y%m%d%H%M%SZ').encode('ascii') @@ -68,10 +70,22 @@ def build_ap_req(username: str, # Encrypt authenticator auth_inner = b'\x30' + asn1_len(auth_body) + auth_body auth_asn1 = b'\x62' + asn1_len(auth_inner) + auth_inner - enc_auth = encrypt_kerberos_rc4(service_session_key, 11, auth_asn1) - etype_asn1 = asn1_tag(0, b'\x02\x01\x17') # RC4-HMAC (23) + if etype == 18: # AES-256 + enc_auth = encrypt_kerberos_aes_cts(service_session_key, 11, auth_asn1) + etype_byte = b'\x12' + elif etype == 17: # AES-128 + enc_auth = encrypt_kerberos_aes_cts(service_session_key, 11, auth_asn1) + etype_byte = b'\x11' + elif etype == 23: # RC4 + enc_auth = encrypt_kerberos_rc4(service_session_key, 11, auth_asn1) + etype_byte = b'\x17' + else: + raise ValueError(f"Invalid E-type: {etype}") + + etype_asn1 = asn1_tag(0, b'\x02\x01' + etype_byte) cipher_asn1 = asn1_tag(2, asn1_ostr(enc_auth)) + enc_body = etype_asn1 + cipher_asn1 enc_part = asn1_tag(4, b'\x30' + asn1_len(enc_body) + enc_body) @@ -87,45 +101,122 @@ def build_ap_req(username: str, return b'\x6e' + asn1_len(inner_pdu) + inner_pdu -def get_active_key(auth_bytes: bytes, - service_session_key: bytes) -> tuple[Optional[bytes], int]: - active_key, seq_number = None, 0 +def _rc4_helper(asn1_data: bytes) -> Tuple[Optional[bytes], Optional[int]]: + active_key = None + seq_number = None + + for i in range(len(asn1_data) - 10): + if asn1_data[i] == 0xa2: + content = peel_tag(asn1_data[i:], 0xa2) + if content: + val_idx = content.find(b'\xa1') + if val_idx != -1: + val_wrap = peel_tag(content[val_idx:], 0xa1) + if val_wrap: + res_key = peel_tag(val_wrap, 0x04) + if res_key and len(res_key) == 16: + active_key = res_key + break + + for i in range(len(asn1_data) - 2): + if asn1_data[i] == 0xa3: + content = peel_tag(asn1_data[i:], 0xa3) + if content: + inner = peel_tag(content, 0x02) + if inner: + seq_number = int.from_bytes(inner, 'big') + break + elif 0 < len(content) <= 8: + seq_number = int.from_bytes(content, 'big') + break - etype_idx = auth_bytes.find(b'\x02\x01\x17') - if etype_idx >= 0: - # Read new session key... - idx_a2 = auth_bytes.find(b'\xa2', etype_idx) - idx_04 = auth_bytes.find(b'\x04', idx_a2) + return active_key, seq_number - pos = idx_04 + 1 - length_byte = auth_bytes[pos] - pos += 1 - if length_byte & 0x80: - n = length_byte & 0x7f - length = int.from_bytes(auth_bytes[pos: pos + n], 'big') - pos += n - else: - length = length_byte - cipher_blob = auth_bytes[pos: pos + length] +def _aes_helper(asn1_data: bytes) -> Tuple[Optional[bytes], Optional[int]]: + active_key = None + seq_number = None + + current = asn1_data + if asn1_data.startswith(b'\x7b'): + current = peel_tag(asn1_data, 0x7b) + if current and current.startswith(b'\x30'): + current = peel_tag(current, 0x30) + + if current: + subkey_cont = peel_tag(current, 0xa2) + if subkey_cont: + k_seq = peel_tag(subkey_cont, 0x30) + if k_seq: + k_val_wrap = peel_tag(k_seq, 0xa1) + if k_val_wrap: + active_key = peel_tag(k_val_wrap, 0x04) - decrypted_raw = \ + seq_cont = peel_tag(current, 0xa3) + if seq_cont: + inner_seq = peel_tag(seq_cont, 0x02) + if inner_seq: + seq_number = int.from_bytes(inner_seq, 'big') + + return active_key, seq_number + + +def get_active_key(auth_bytes: bytes, + service_session_key: bytes, + etype: int) -> Tuple[Optional[bytes], Optional[int]]: + active_key, seq_number = None, None + kerberos_data = auth_bytes + + if auth_bytes.startswith(b'\xa1'): + idx_a2 = auth_bytes.find(b'\xa2') + if idx_a2 == -1: + raise NoNewActiveKey() + + idx_04 = auth_bytes.find(b'\x04', idx_a2) + if idx_04 != -1: + pos = idx_04 + 1 + lb = auth_bytes[pos] + pos += 1 + if lb & 0x80: + n = lb & 0x7f + length = int.from_bytes(auth_bytes[pos:pos+n], 'big') + pos += n + else: + length = lb + kerberos_data = auth_bytes[pos: pos + length] + else: + raise NoNewActiveKey() + + etype_idx = kerberos_data.find(bytes([0x02, 0x01, etype])) + if etype_idx == -1: + return None, None + + idx_04_cipher = kerberos_data.find(b'\x04', etype_idx) + if idx_04_cipher == -1: + return None, None + + pos = idx_04_cipher + 1 + lb = kerberos_data[pos] + pos += 1 + if lb & 0x80: + n = lb & 0x7f + c_length = int.from_bytes(kerberos_data[pos: pos + n], 'big') + pos += n + else: + c_length = lb + cipher_blob = kerberos_data[pos: pos + c_length] + + if etype == 23: # RC4 + decrypted = \ decrypt_kerberos_rc4(service_session_key, 12, cipher_blob) - asn1_data = decrypted_raw[8:] # Skip 8 bytes confounder - - KEY_MARKER = b'\xa1\x12\x04\x10' - key_idx = asn1_data.find(KEY_MARKER) - if key_idx != -1: - active_key = asn1_data[key_idx + 4: key_idx + 4 + 16] - - SEQ_MARKER = b'\xa3' - seq_idx = asn1_data.find(SEQ_MARKER) - if seq_idx != -1: - int_tag_idx = asn1_data.find(b'\x02', seq_idx) - if int_tag_idx != -1 and int_tag_idx < seq_idx + 4: - int_len = asn1_data[int_tag_idx + 1] - offset = int_tag_idx + 2 - seq_bytes = asn1_data[offset: offset + int_len] - seq_number = int.from_bytes(seq_bytes, 'big') + # Skip 8 bytes confounder + active_key, seq_number = _rc4_helper(decrypted[8:]) + elif etype in (17, 18): # AES + decrypted = \ + decrypt_kerberos_aes_cts(service_session_key, 12, cipher_blob) + # Skip 16 bytes confounder + active_key, seq_number = _aes_helper(decrypted[16:]) + else: + raise ValueError(f"Invalid E-type: {etype}") return active_key, seq_number diff --git a/aiowmi/kerberos/as_req.py b/aiowmi/kerberos/as_req.py index b27e0ca..695d409 100644 --- a/aiowmi/kerberos/as_req.py +++ b/aiowmi/kerberos/as_req.py @@ -61,6 +61,9 @@ def build_as_req(username: str, domain: str, pa_enc: bytes = b'') -> bytes: # Renewable time: 1 day (must be >= till) rtime_time = (now + timedelta(days=1)).strftime("%Y%m%d%H%M%SZ").encode() + # Etype list: RC4 (23), AES128 (17), AES256 (18) + etypes = asn1_int(18) + # kdc-options: Forwardable, Proxiable, Renewable, Canonicalize kdc_options = b'\x03\x05\x00\x50\x80\x00\x00' req_body_fields = ( @@ -72,8 +75,7 @@ def build_as_req(username: str, domain: str, pa_enc: bytes = b'') -> bytes: asn1_tag(5, asn1_gt(till_time)) + asn1_tag(6, asn1_gt(rtime_time)) + asn1_tag(7, asn1_int(nonce)) + - # etype: aes256-cts-hmac-sha1-96 (18) - asn1_tag(8, asn1_seq(asn1_int(18))) + asn1_tag(8, asn1_seq(etypes)) ) req_body_sequence = asn1_seq(req_body_fields) @@ -91,12 +93,14 @@ def build_as_req(username: str, domain: str, pa_enc: bytes = b'') -> bytes: return final_as_req -def build_full_as_req(username: str, domain: str, base_key: bytes): +def build_full_as_req(username: str, domain: str, base_key: bytes, etype: int): now = datetime.now(timezone.utc) + micro = now.microsecond + ts_str = now.strftime("%Y%m%d%H%M%SZ").encode() original_plain_body = ( asn1_tag(0, asn1_gt(ts_str)) + # [0] pausec (GeneralizedTime) - asn1_tag(1, asn1_int(0)) # [1] pusec (Microseconds, 0) + asn1_tag(1, asn1_int(micro)) # [1] pusec (Microseconds) ) original_plain = asn1_seq(original_plain_body) @@ -105,13 +109,14 @@ def build_full_as_req(username: str, domain: str, base_key: bytes): ke = derive_key(base_key, 1, 0xAA) ki = derive_key(base_key, 1, 0x55) + signature = hmac.new(ki, full_plain, hashlib.sha1).digest()[:12] cipher_only = aes_cts_encrypt(ke, full_plain) final_payload = cipher_only + signature enc_data_content = ( - asn1_tag(0, asn1_int(18)) + # etype 18 (AES-256) + asn1_tag(0, asn1_int(etype)) + # etype 17 (AES128) or 18 (AES256) asn1_tag(2, asn1_ostr(final_payload)) ) @@ -144,6 +149,9 @@ def build_full_as_req(username: str, domain: str, base_key: bytes): asn1_tag(1, asn1_seq(asn1_gs(b"krbtgt") + asn1_gs(domcaps.encode()))) ) + # Etype list: RC4 (23), AES128 (17), AES256 (18) + etypes = asn1_int(18) + req_body_content = ( asn1_tag(0, b'\x03\x05\x00\x50\x80\x00\x00') + # kdc-options asn1_tag(1, asn1_seq(cname_content)) + @@ -152,7 +160,7 @@ def build_full_as_req(username: str, domain: str, base_key: bytes): asn1_tag(5, asn1_gt(till)) + asn1_tag(6, asn1_gt(till)) + asn1_tag(7, asn1_int(nonce)) + - asn1_tag(8, asn1_seq(asn1_int(18))) # etype list + asn1_tag(8, asn1_seq(etypes)) ) as_req_content = ( diff --git a/aiowmi/kerberos/asn1.py b/aiowmi/kerberos/asn1.py index 0ac9ab2..fdc2bf8 100644 --- a/aiowmi/kerberos/asn1.py +++ b/aiowmi/kerberos/asn1.py @@ -1,5 +1,5 @@ import struct -from typing import Union +from typing import Union, Tuple def asn1_tag(tag_num: int, content: bytes) -> bytes: @@ -46,7 +46,7 @@ def asn1_ostr(val: bytes) -> bytes: return b'\x04' + asn1_len(val) + val -def get_asn1_len(data: bytes, pos: int) -> tuple[int, int]: +def get_asn1_len(data: bytes, pos: int) -> Tuple[int, int]: """Read ASN.1 lengte from data.""" b = data[pos] if b < 128: diff --git a/aiowmi/kerberos/cache.py b/aiowmi/kerberos/cache.py index a350be8..63bebb4 100644 --- a/aiowmi/kerberos/cache.py +++ b/aiowmi/kerberos/cache.py @@ -1,6 +1,6 @@ import pickle from logging import Logger -from typing import Optional +from typing import Optional, Tuple class KerberosCache: @@ -16,27 +16,28 @@ def __init__(self, file_path: Optional[str] = None): If None, memory is used (must keep the KerberosCache alive) """ self._file_path = file_path - self._tgt: Optional[tuple[bytes, bytes]] = None - self._tgs: Optional[tuple[bytes, bytes]] = None + self._tgt: Optional[Tuple[bytes, bytes]] = None + self._tgs: Optional[Tuple[bytes, bytes, int]] = None - def open(self, logger: Logger) -> tuple[ - Optional[tuple[bytes, bytes]], - Optional[tuple[bytes, bytes]]]: + def open(self, logger: Logger) -> Tuple[ + Optional[Tuple[bytes, bytes]], + Optional[Tuple[bytes, bytes, int]]]: if self._file_path is not None and \ (self._tgs is None or self._tgt is None): try: with open(self._file_path, 'rb') as fp: dump = pickle.load(fp) self._tgt = dump[0], dump[1] - self._tgs = dump[2], dump[3] + self._tgs = dump[2], dump[3], dump[4] except Exception: logger.warning(f'Failed to load from: {self._file_path}') + self._tgt, self._tgs = None, None return self._tgt, self._tgs def write(self, - tgt: tuple[bytes, bytes], - tgs: tuple[bytes, bytes], + tgt: Tuple[bytes, bytes], + tgs: Tuple[bytes, bytes, int], logger: Logger): self._tgt = tgt self._tgs = tgs diff --git a/aiowmi/kerberos/const.py b/aiowmi/kerberos/const.py index 6702e1d..2da134d 100644 --- a/aiowmi/kerberos/const.py +++ b/aiowmi/kerberos/const.py @@ -1,3 +1,3 @@ -OID_KERBEROS_V5 = b'\x06\t\x2a\x86\x48\x86\xf7\x12\x01\x02\x02' -OID_MS_LEGACY_KRB = b'\x06\t\x2a\x86\x48\x82\xf7\x12\x01\x02\x02' OID_SPNEGO = b'\x06\x06\x2b\x06\x01\x05\x05\x02' +OID_IETF_KRB5 = b'\x06\x09\x2a\x86\x48\x82\xf7\x12\x01\x02\x02' +OID_MS_KRB5 = b'\x06\x09\x2a\x86\x48\x86\xf7\x12\x01\x02\x02' diff --git a/aiowmi/kerberos/gss.py b/aiowmi/kerberos/gss.py index 17521ac..6f8e9e4 100644 --- a/aiowmi/kerberos/gss.py +++ b/aiowmi/kerberos/gss.py @@ -1,7 +1,9 @@ import struct +from typing import Tuple from Crypto.Hash import HMAC, MD5 from Crypto.Cipher import ARC4 from ..tools import get_random_bytes +from .tools import encrypt_kerberos_aes_cts, decrypt_kerberos_aes_cts GSS_WRAP_HEADER = b'\x60\x2b\x06\x09\x2a\x86\x48\x86\xf7\x12\x01\x02\x02' @@ -93,3 +95,68 @@ def gss_unwrap_rc4(session_key: bytes, raise Exception(f"Integrity Check Failed! Server Seq was: {svr_seq}") return data + + +def gss_wrap_aes(session_key: bytes, + data: bytes, + seq_num: int) -> Tuple[bytes, bytes]: + pad = (16 - (len(data) % 16)) & 15 + pad_str = b'\xFF' * pad + + rcc = 0 + header_for_hash = ( + b'\x05\x04\x06\xff' + + struct.pack('>H', pad) + + struct.pack('>H', rcc) + + struct.pack('>Q', seq_num) + ) + + plaintext = data + pad_str + header_for_hash + raw_cipher = encrypt_kerberos_aes_cts(session_key, 24, plaintext) + + def rotate(bytes_data: bytes, n: int) -> bytes: + n %= len(bytes_data) + left = len(bytes_data) - n + return bytes_data[left:] + bytes_data[:left] + + rrc = 28 + total_rotate = rrc + pad + cipher_rotated = rotate(raw_cipher, total_rotate) + + wire_header = ( + b'\x05\x04\x06\xff' + + struct.pack('>H', pad) + + struct.pack('>H', rrc) + + struct.pack('>Q', seq_num) + ) + split_offset = 16 + rrc + pad + + ret2 = wire_header + cipher_rotated[:split_offset] + ret1 = cipher_rotated[split_offset:] + + return ret1, ret2 + + +def gss_unwrap_aes(session_key: bytes, cipher_text: bytes, + auth_data: bytes) -> bytes: + header = auth_data[:16] + pad = struct.unpack('>H', header[4:6])[0] # EC + rrc = struct.unpack('>H', header[6:8])[0] # RRC + + cipher_from_trailer = auth_data[16:] + + rotated_blob = cipher_from_trailer + cipher_text + + def unrotate(data, n): + if not data: + return data + n %= len(data) + return data[n:] + data[:n] + + full_cipher_blob = unrotate(rotated_blob, rrc + pad) + + decrypted_payload = \ + decrypt_kerberos_aes_cts(session_key, 22, full_cipher_blob) + actual_data = decrypted_payload[16: -(pad + 16)] + + return actual_data diff --git a/aiowmi/kerberos/krb5_pdu.py b/aiowmi/kerberos/krb5_pdu.py index 6c42258..aa9de65 100644 --- a/aiowmi/kerberos/krb5_pdu.py +++ b/aiowmi/kerberos/krb5_pdu.py @@ -1,12 +1,12 @@ import datetime import struct from .asn1 import asn1_len, asn1_seq, asn1_tag, asn1_int, asn1_gt, asn1_ostr -from .tools import encrypt_kerberos_rc4 +from .tools import encrypt_kerberos_rc4, encrypt_kerberos_aes_cts from ..dcom_const import NDR_TransferSyntaxIdentifier from ..rpc.const import RPC_C_AUTHN_GSS_NEGOTIATE -def get_neg_token(service_session_key: bytes, seq_number: int): +def get_neg_token(service_session_key: bytes, seq_number: int, etype: int): now = datetime.datetime.now(datetime.timezone.utc) timestamp = now.strftime('%Y%m%d%H%M%SZ').encode('ascii') cusec = now.microsecond @@ -14,26 +14,38 @@ def get_neg_token(service_session_key: bytes, seq_number: int): enc_ap_rep_body = ( asn1_tag(0, asn1_gt(timestamp)) + # [0] ctime asn1_tag(1, asn1_int(cusec)) + # [1] cusec - asn1_tag(3, asn1_int(seq_number)) + asn1_tag(3, asn1_int(seq_number)) # [3] seq-number ) - inner_seq = asn1_seq(enc_ap_rep_body) - enc_ap_rep = b'\x7b' + asn1_len(inner_seq) + inner_seq - encrypted_data = encrypt_kerberos_rc4(service_session_key, 12, enc_ap_rep) + inner_seq = b'\x30' + asn1_len(enc_ap_rep_body) + enc_ap_rep_body + plaintext = b'\x7b' + asn1_len(inner_seq) + inner_seq + + if etype in (17, 18): # AES-128 / AES-256 + enc_data = encrypt_kerberos_aes_cts(service_session_key, 12, plaintext) + current_etype = etype + else: # RC4 (23) + enc_data = encrypt_kerberos_rc4(service_session_key, 12, plaintext) + current_etype = 23 + enc_part_content = ( - asn1_tag(0, asn1_int(23)) + # etype 23 (RC4) - asn1_tag(2, asn1_ostr(encrypted_data)) # cipher + asn1_tag(0, asn1_int(current_etype)) + + asn1_tag(2, asn1_ostr(enc_data)) ) ap_rep_body = ( asn1_tag(0, asn1_int(5)) + # pvno asn1_tag(1, asn1_int(15)) + # msg-type 15 (AP-REP) - asn1_tag(2, asn1_seq(enc_part_content)) # enc-part + asn1_tag(2, b'\x30' + asn1_len(enc_part_content) + enc_part_content) ) - ap_rep = b'\x6f' + asn1_len(asn1_seq(ap_rep_body)) + asn1_seq(ap_rep_body) - response_token = asn1_tag(2, asn1_ostr(ap_rep)) - auth_neg_token = asn1_tag(1, asn1_seq(response_token)) + ap_rep_wrap = b'\x30' + asn1_len(ap_rep_body) + ap_rep_body + ap_rep_full = b'\x6f' + asn1_len(ap_rep_wrap) + ap_rep_wrap + + # MechToken [2] -> Octet String -> AP-REP + resp_token = asn1_tag(2, asn1_ostr(ap_rep_full)) + + # NegTokenResp [1] -> Sequence -> MechToken + auth_neg_token = asn1_tag(1, b'\x30' + asn1_len(resp_token) + resp_token) return auth_neg_token diff --git a/aiowmi/kerberos/tgs.py b/aiowmi/kerberos/tgs.py index 66bbcbd..2d1b238 100644 --- a/aiowmi/kerberos/tgs.py +++ b/aiowmi/kerberos/tgs.py @@ -1,4 +1,5 @@ import random +from typing import Tuple from Crypto.Cipher import AES from datetime import datetime, timezone, timedelta from pyasn1.codec.der import decoder, encoder @@ -105,7 +106,7 @@ def build_tgs_req(username: str, domain: str, session_key: bytes, ticket_bytes: bytes, - target_service: tuple[str, str]): + target_service: Tuple[str, str]): now = datetime.now(timezone.utc) ts_str = now.strftime("%Y%m%d%H%M%SZ").encode() till_ts = (now + timedelta(hours=8)).strftime("%Y%m%d%H%M%SZ").encode() @@ -160,8 +161,8 @@ def build_tgs_req(username: str, )) ) - # Etype list: RC4, AES128, DES, AES256 - etype_list = asn1_int(23) + asn1_int(16) + asn1_int(3) + asn1_int(18) + # Etype list: RC4 (23), AES128 (17), AES256 (18) + etypes = asn1_int(23) + asn1_int(17) + asn1_int(18) nonce = random.getrandbits(31) req_body_content = ( @@ -170,7 +171,7 @@ def build_tgs_req(username: str, asn1_tag(3, asn1_seq(sname_inner)) + # sname asn1_tag(5, asn1_gt(till_ts)) + # Till asn1_tag(7, asn1_int(nonce)) + # Nonce - asn1_tag(8, asn1_seq(etype_list)) # Etypes + asn1_tag(8, asn1_seq(etypes)) # Etypes ) req_body_field = asn1_tag(4, asn1_seq(req_body_content)) @@ -217,7 +218,7 @@ def extract_ticket(data): def get_service_key(resp_bytes: bytes, - session_key: bytes) -> tuple[bytes, bytes]: + session_key: bytes) -> Tuple[bytes, bytes, int]: raw_obj, _ = decoder.decode(resp_bytes) enc_part = raw_obj.getComponentByPosition(5) @@ -236,26 +237,24 @@ def get_service_key(resp_bytes: bytes, inner_obj, _ = decoder.decode(payload) key_sequence = inner_obj.getComponentByPosition(0) + service_etype = int(key_sequence.getComponentByPosition(0)) service_session_key = bytes(key_sequence.getComponentByPosition(1)) ticket = raw_obj.getComponentByPosition(4) ticket_bytes = encoder.encode(ticket) - WRAPPERS = [0xa3, 0xa4, 0xa5] + if ticket_bytes[0] != 0x61: + first_len_byte = ticket_bytes[1] + num_len_bytes = first_len_byte & 0x7f if first_len_byte & 0x80 else 0 + ticket_bytes = ticket_bytes[2 + num_len_bytes:] - if ticket_bytes[0] in WRAPPERS: - try: - ticket_start = ticket_bytes.index(b'\x61', 0, 10) - ticket_bytes = ticket_bytes[ticket_start:] - except ValueError: - pass - - return ticket_bytes, service_session_key + return ticket_bytes, service_session_key, service_etype async def get_tgs(username: str, domain: str, host: str, as_rep_bytes: bytes, base_key: bytes, - kdc_host: str, kdc_port: int = 88) -> tuple[bytes, bytes]: + kdc_host: str, + kdc_port: int = 88) -> Tuple[bytes, bytes, int]: tgs_session_key = get_session_key(as_rep_bytes, base_key) tgs_ticket = extract_ticket(as_rep_bytes) tgs_req = build_tgs_req(username, @@ -265,5 +264,5 @@ async def get_tgs(username: str, domain: str, host: str, ("host", host)) as_res_bytes = await send_kerberos_packet(tgs_req, kdc_host, kdc_port) parse_krb_error(as_res_bytes) - ticket, service_key = get_service_key(as_res_bytes, tgs_session_key) - return ticket, service_key + ticket, service_key, etype = get_service_key(as_res_bytes, tgs_session_key) + return ticket, service_key, etype diff --git a/aiowmi/kerberos/tgt.py b/aiowmi/kerberos/tgt.py index d11afc1..1750242 100644 --- a/aiowmi/kerberos/tgt.py +++ b/aiowmi/kerberos/tgt.py @@ -1,44 +1,63 @@ -import hashlib +from typing import Tuple from Crypto.Protocol.KDF import PBKDF2 from Crypto.Cipher import AES from Crypto.Hash import SHA1 from .as_req import build_as_req, build_full_as_req from .kdc import send_kerberos_packet -from .asn1 import get_asn1_len -from .tools import parse_krb_error +from .tools import parse_krb_error, peel_tag -def extract_salt_and_etype(error_data: bytes) -> tuple[str, int]: +def extract_salt_and_etype(error_data: bytes) -> Tuple[str, int]: if error_data[0] != 0x7e: raise ValueError("Invalid KRB-ERROR packet") e_data_idx = error_data.find(b'\xac') if e_data_idx == -1: - raise ValueError("e-data (Tag 12) not found") + raise ValueError("e-data (Tag 12) not found in KRB-ERROR") - salt_tag_idx = error_data.find(b'\x1b', e_data_idx) + etype_18_marker = b'\x02\x01\x12' + marker_idx = error_data.find(etype_18_marker, e_data_idx) + + etype = 18 + if marker_idx == -1: + etype_17_marker = b'\x02\x01\x11' + marker_idx = error_data.find(etype_17_marker, e_data_idx) + etype = 17 + + if marker_idx == -1: + raise ValueError("No AES (17/18) etype found in e-data") + + salt_tag_idx = error_data.find(b'\xa1', marker_idx) if salt_tag_idx == -1: - # Default to no salt/empty string if not found - salt = "" + raise ValueError(f"Salt tag (0xa1) not found for etype {etype}") + + salt_start_idx = -1 + for tag in [b'\x1b', b'\x04', b'\x1d']: + tag_pos = error_data.find(tag, salt_tag_idx, salt_tag_idx + 15) + if tag_pos != -1: + salt_start_idx = tag_pos + break + + if salt_start_idx == -1: + raise ValueError( + "Could not find string tag (1b/04) within salt container") + + pos = salt_start_idx + 1 + length_byte = error_data[pos] + pos += 1 + + if length_byte & 0x80: + n_bytes = length_byte & 0x7f + salt_len = int.from_bytes(error_data[pos: pos + n_bytes], 'big') + pos += n_bytes else: - salt_len, salt_len_size = get_asn1_len(error_data, salt_tag_idx + 1) - salt_start = salt_tag_idx + 1 + salt_len_size - salt = error_data[salt_start: salt_start + salt_len].decode('utf-8') - - etype = 18 # Default to AES256 if we can't find it - etype_search_start = error_data.find( - b'\xa0', - e_data_idx, - salt_tag_idx if salt_tag_idx != -1 else None) - - if etype_search_start != -1: - # The etype is an integer inside context tag 0xa0 - val_idx = error_data.find(b'\x02', etype_search_start) - if val_idx != -1: - e_len = error_data[val_idx + 1] - offset = val_idx + 2 - etype = int.from_bytes(error_data[offset: offset + e_len], 'big') + salt_len = length_byte + + salt_bytes = error_data[pos: pos + salt_len] + if not salt_bytes: + raise ValueError("Extracted salt is empty") + salt = salt_bytes.decode('utf-8', errors='replace') return salt, etype @@ -46,13 +65,13 @@ def nfold_for_kerberos(): return b'\x6b\x65\x72\x62\x65\x72\x6f\x73\x7b\x9b\x5b\x2b\x93\x13\x2b\x93' -def aes_string_to_key(password, salt, key_len=32): +def aes_string_to_key(password: str, salt: str, key_len: int = 32): """ Kerberos String-to-Key for AES (RFC 3962) key_len should be 16 for AES-128 or 32 for AES-256 """ tkey = PBKDF2( - password.encode(), + password, salt.encode(), dkLen=key_len, count=4096, @@ -78,14 +97,15 @@ def aes_string_to_key(password, salt, key_len=32): async def get_tgt(username: str, password: str, domain: str, - kdc_host: str, kdc_port: int = 88) -> tuple[bytes, bytes]: + kdc_host: str, kdc_port: int = 88) -> Tuple[bytes, bytes]: as_req = build_as_req(username, domain) resp = await send_kerberos_packet(as_req, kdc_host, kdc_port) parse_krb_error(resp) salt, etype = extract_salt_and_etype(resp) - key_len = 16 if etype == 17 else 32 - base_key = aes_string_to_key(password, salt, key_len) - full_as_req = build_full_as_req(username, domain, base_key) + if etype != 18: + raise ValueError('Only AES256 encryption supported for negotiation') + base_key = aes_string_to_key(password, salt) + full_as_req = build_full_as_req(username, domain, base_key, etype) as_res_bytes = await send_kerberos_packet(full_as_req, kdc_host, kdc_port) parse_krb_error(as_res_bytes) return as_res_bytes, base_key diff --git a/aiowmi/kerberos/tools.py b/aiowmi/kerberos/tools.py index eec04d7..77311a4 100644 --- a/aiowmi/kerberos/tools.py +++ b/aiowmi/kerberos/tools.py @@ -4,51 +4,47 @@ import hmac import math import struct +from typing import Optional from functools import reduce from pyasn1.codec.der import decoder, encoder from Crypto.Cipher import AES from Crypto.Hash import HMAC, SHA1, MD5 from .rc4 import RC4 -from .gss import gss_wrap_rc4, gss_unwrap_rc4 from ..exceptions import KerberosErr from ..tools import get_random_bytes def _nfold(ba, nbytes): def rotate_right(ba, nbits): - ba = bytearray(ba) n = len(ba) nbits %= (n * 8) - # Rotate bytes - for _ in range(nbits // 8): - ba = bytearray([ba[-1]]) + ba[:-1] - # Rotate remaining bits - nbits %= 8 - if nbits > 0: - last_bit = 0 - for i in range(len(ba)): - new_last_bit = ba[i] & ((1 << nbits) - 1) - ba[i] = (ba[i] >> nbits) | (last_bit << (8 - nbits)) - last_bit = new_last_bit - ba[0] |= (last_bit << (8 - nbits)) - return ba + if nbits == 0: + return bytearray(ba) + + val = int.from_bytes(ba, 'big') + + mask = (1 << (n * 8)) - 1 + rotated = ((val >> nbits) | (val << (n * 8 - nbits))) & mask + + return bytearray(rotated.to_bytes(n, 'big')) def add_ones_complement(ba1, ba2): n = len(ba1) - res = [0] * n + res = bytearray(n) carry = 0 for i in range(n-1, -1, -1): s = ba1[i] + ba2[i] + carry res[i] = s & 0xff carry = s >> 8 - pos = n - 1 - while carry and pos >= 0: - s = res[pos] + carry - res[pos] = s & 0xff - carry = s >> 8 - pos -= 1 - return bytearray(res) + while carry: + for i in range(n-1, -1, -1): + s = res[i] + carry + res[i] = s & 0xff + carry = s >> 8 + if not carry: + break + return res slen = len(ba) @@ -58,7 +54,7 @@ def lcm(a, b): lcm_val = lcm(slen, nbytes) big_str = bytearray() curr = bytearray(ba) - for i in range(lcm_val // slen): + for _ in range(lcm_val // slen): big_str += curr curr = rotate_right(curr, 13) @@ -66,19 +62,21 @@ def lcm(a, b): return bytes(reduce(add_ones_complement, parts)) -def derive_key(base_key, usage_int, payload_byte=None): - constant = struct.pack('>I', usage_int) - if payload_byte is not None: - constant += bytes([payload_byte]) +def derive_key(base_key: bytes, usage: int, payload: int) -> bytes: + key_len = len(base_key) # 16 or 32 + + constant = struct.pack('>I', usage) + constant += bytes([payload]) nfolded = _nfold(constant, 16) - cipher1 = AES.new(base_key, AES.MODE_ECB) - b1 = cipher1.encrypt(nfolded) + cipher = AES.new(base_key, AES.MODE_ECB) + b1 = cipher.encrypt(nfolded) - cipher2 = AES.new(base_key, AES.MODE_ECB) - b2 = cipher2.encrypt(b1) + if key_len == 16: + return b1 + b2 = cipher.encrypt(b1) return b1 + b2 @@ -129,40 +127,31 @@ def decrypt_kerberos_aes_cts(key: bytes, usage: int, cipher: bytes) -> bytes: if n == block_size: # Simple one-block case plaintext_with_confounder = aes_ecb.decrypt(ciphertext) - elif n % block_size == 0: - # Standard CBC - cipher_cbc = AES.new(ke, AES.MODE_CBC, b'\x00' * block_size) - plaintext_with_confounder = cipher_cbc.decrypt(ciphertext) else: - # Manual CTS Decryption (NIST CS3) - m = (n // block_size) * block_size - last_len = n % block_size + # NIST CTS Decryption + m = ((n - block_size - 1) // block_size) * block_size + + p_start = b"" + iv = b'\x00' * block_size + if m > 0: + cipher_cbc = AES.new(ke, AES.MODE_CBC, iv) + p_start = cipher_cbc.decrypt(ciphertext[:m]) + iv = ciphertext[m-block_size:m] - # All blocks except last two - prev_blocks = ciphertext[:m-block_size] - cn_minus_1 = ciphertext[m-block_size:m] - cn = ciphertext[m:] + cn_minus_1 = ciphertext[m: m + block_size] + cn = ciphertext[m + block_size:] - # Decrypt cn_minus_1 to get the 'stolen' padding tmp = aes_ecb.decrypt(cn_minus_1) - pn = bytes(a ^ b for a, b in zip(tmp[:last_len], cn)) - - # Reconstruct the full last block - last_block_full = cn + tmp[last_len:] - pn_minus_1_dec = aes_ecb.decrypt(last_block_full) - - # IV for the second-to-last block - iv = b'\x00' * block_size \ - if m == block_size \ - else ciphertext[m-block_size*2:m-block_size] - pn_minus_1 = bytes(a ^ b for a, b in zip(pn_minus_1_dec, iv)) - - if m > block_size: - cipher_cbc = AES.new(ke, AES.MODE_CBC, b'\x00' * block_size) - p_start = cipher_cbc.decrypt(prev_blocks) - plaintext_with_confounder = p_start + pn_minus_1 + pn - else: - plaintext_with_confounder = pn_minus_1 + pn + + pn = bytes(a ^ b for a, b in zip(tmp[:len(cn)], cn)) + + last_block_full = cn + tmp[len(cn):] + + pn_minus_1 = bytes(a ^ b + for a, b + in zip(aes_ecb.decrypt(last_block_full), iv)) + + plaintext_with_confounder = p_start + pn_minus_1 + pn h = HMAC.new(ki, plaintext_with_confounder, SHA1) if h.digest()[:12] != expected_hmac: @@ -173,34 +162,33 @@ def decrypt_kerberos_aes_cts(key: bytes, usage: int, cipher: bytes) -> bytes: def encrypt_kerberos_aes_cts(session_key: bytes, usage: int, - plain_text: bytes): + plain_text: bytes, + confounder: Optional[bytes] = None): ki = derive_key(session_key, usage, 0x55) ke = derive_key(session_key, usage, 0xAA) - confounder = get_random_bytes(16) - basic_plaintext = confounder + plain_text + if confounder is None: + confounder = get_random_bytes(16) + plaintext = confounder + plain_text - h = HMAC.new(ki, basic_plaintext, SHA1) + h = HMAC.new(ki, plaintext, SHA1) checksum = h.digest()[:12] aes = AES.new(ke, AES.MODE_CBC, b'\x00' * 16) - n = len(basic_plaintext) - if n % 16 == 0: - final_ctext = aes.encrypt(basic_plaintext) - else: - pad_len = 16 - (n % 16) - padded_data = basic_plaintext + b'\x00' * pad_len - ctext = aes.encrypt(padded_data) + pad_len = (16 - (len(plaintext) % 16)) % 16 + padded_data = plaintext + (b'\x00' * pad_len) + + ctext = aes.encrypt(padded_data) - last_full_block = ctext[-32:-16] - truncated_block = ctext[-16:] + last_full_block = ctext[-32:-16] + truncated_block = ctext[-16:] - final_ctext = ( - ctext[:-32] + - truncated_block + - last_full_block[:16-pad_len] - ) + final_ctext = ( + ctext[:-32] + + truncated_block + + last_full_block[:16-pad_len] + ) return final_ctext + checksum @@ -252,42 +240,6 @@ def read_session_key(data: bytes) -> bytes: return session_key -def seal_func_kerberos(session_key: bytes): - def _kerberos_sealer( - flags: int, - seq_num: int, - message_to_sign: bytes, - message_to_encrypt: bytes, - session_key: bytes) -> tuple[bytes, bytes]: - return gss_wrap_rc4(session_key, - message_to_encrypt, - seq_num) - return functools.partial(_kerberos_sealer, session_key=session_key) - - -def gss_unwrap_kerberos(session_key: bytes): - def _kerberos_unwrap( - cipher_text: bytes, - auth_data: bytes, - session_key: bytes) -> bytes: - return gss_unwrap_rc4(session_key, - cipher_text, - auth_data) - return functools.partial(_kerberos_unwrap, session_key=session_key) - - -def sign_func_kerberos(session_key: bytes): - def _kerberos_signer( - flags: int, - seq_num: int, - message_to_sign: bytes, - session_key: bytes) -> tuple[bytes, bytes]: - return gss_wrap_rc4(session_key, - message_to_sign, - seq_num) - return functools.partial(_kerberos_signer, session_key=session_key) - - KDC_ERR_PREAUTH_REQUIRED = 25 KRB_ERRORS = { @@ -402,3 +354,27 @@ def parse_krb_error(data: bytes): raise KerberosErr( KRB_ERRORS.get(code, f"Unknown Kerberos Error: {code}") ) + + +def peel_tag(data: bytes, target_tag: int) -> Optional[bytes]: + if not data: + return None + p = 0 + while p < len(data): + tag = data[p] + p += 1 + if p >= len(data): + break + lb = data[p] + p += 1 + if lb & 0x80: + n_lb = lb & 0x7f + c_len = int.from_bytes(data[p: p+n_lb], 'big') + p += n_lb + else: + c_len = lb + if tag == target_tag: + return data[p: p + c_len] + p += c_len + + return None diff --git a/aiowmi/kerberos/wrappers.py b/aiowmi/kerberos/wrappers.py new file mode 100644 index 0000000..0ecc687 --- /dev/null +++ b/aiowmi/kerberos/wrappers.py @@ -0,0 +1,84 @@ +import functools +from typing import Tuple +from .gss import gss_wrap_rc4, gss_wrap_aes, gss_unwrap_rc4, gss_unwrap_aes + + +def seal_func_kerberos(session_key: bytes, etype: int): + def _sealer_rc4( + flags: int, + seq_num: int, + message_to_sign: bytes, + message_to_encrypt: bytes, + session_key: bytes) -> Tuple[bytes, bytes]: + return gss_wrap_rc4(session_key, + message_to_encrypt, + seq_num) + + def _sealer_aes( + flags: int, + seq_num: int, + message_to_sign: bytes, + message_to_encrypt: bytes, + session_key: bytes) -> Tuple[bytes, bytes]: + return gss_wrap_aes(session_key, + message_to_encrypt, + seq_num) + if etype in (17, 18): + sealer = _sealer_aes + elif etype == 23: + sealer = _sealer_rc4 + else: + raise ValueError(f"Invalid E-type: {etype}") + return functools.partial(sealer, session_key=session_key) + + +def gss_unwrap_kerberos(session_key: bytes, etype: int): + def _unwrap_rc4( + cipher_text: bytes, + auth_data: bytes, + session_key: bytes) -> bytes: + return gss_unwrap_rc4(session_key, + cipher_text, + auth_data) + + def _unwrap_aes( + cipher_text: bytes, + auth_data: bytes, + session_key: bytes) -> bytes: + return gss_unwrap_aes(session_key, + cipher_text, + auth_data) + if etype in (17, 18): + unwrapper = _unwrap_aes + elif etype == 23: + unwrapper = _unwrap_rc4 + else: + raise ValueError(f"Invalid E-type: {etype}") + return functools.partial(unwrapper, session_key=session_key) + + +def sign_func_kerberos(session_key: bytes, etype: int): + def _signer_rc4( + flags: int, + seq_num: int, + message_to_sign: bytes, + session_key: bytes) -> Tuple[bytes, bytes]: + return gss_wrap_rc4(session_key, + message_to_sign, + seq_num) + + def _signer_aes( + flags: int, + seq_num: int, + message_to_sign: bytes, + session_key: bytes) -> Tuple[bytes, bytes]: + return gss_wrap_aes(session_key, + message_to_sign, + seq_num) + if etype in (17, 18): + signer = _signer_aes + elif etype == 23: + signer = _signer_rc4 + else: + raise ValueError(f"Invalid E-type: {etype}") + return functools.partial(signer, session_key=session_key)