class TestServer(pyrad.server.Server):
def _HandleAuthPacket(self, pkt):
pyrad.server.Server._HandleAuthPacket(self, pkt)
- eap = ""
+ eap = b''
for p in pkt[79]:
eap += p
eap_req = self.eap_handler(self.ctx, eap)
hmac_obj.update(struct.pack("B", reply.id))
# reply attributes
- reply.AddAttribute("Message-Authenticator",
- "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00")
+ reply.AddAttribute("Message-Authenticator", 16*b'\x00')
attrs = reply._PktEncodeAttributes()
# Length
srv = TestServer(dict=pyrad.dictionary.Dictionary("dictionary.radius"),
authport=18138, acctport=18139)
srv.hosts["127.0.0.1"] = pyrad.server.RemoteHost("127.0.0.1",
- "radius",
+ b"radius",
"localhost")
srv.BindToAddress("")
t_stop = threading.Event()
"""EAP protocol tests"""
check_eap_capa(dev[0], "MD5")
def eap_handler(ctx, req):
- logger.info("eap_handler - RX " + req.encode("hex"))
+ logger.info("eap_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
def test_eap_proto_notification_errors(dev, apdev):
"""EAP Notification errors"""
def eap_handler(ctx, req):
- logger.info("eap_handler - RX " + req.encode("hex"))
+ logger.info("eap_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
EAP_SAKE_AT_RAND_S, 18, 0, 0, 0, 0)
def sake_handler(ctx, req):
- logger.info("sake_handler - RX " + req.encode("hex"))
+ logger.info("sake_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] += 1
def test_eap_proto_sake_errors2(dev, apdev):
"""EAP-SAKE protocol tests (2)"""
def sake_handler(ctx, req):
- logger.info("sake_handler - RX " + req.encode("hex"))
+ logger.info("sake_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] += 1
"""EAP-LEAP protocol tests"""
check_eap_capa(dev[0], "LEAP")
def leap_handler(ctx, req):
- logger.info("leap_handler - RX " + req.encode("hex"))
+ logger.info("leap_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
check_eap_capa(dev[0], "LEAP")
def leap_handler2(ctx, req):
- logger.info("leap_handler2 - RX " + req.encode("hex"))
+ logger.info("leap_handler2 - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
check_eap_capa(dev[0], "MD5")
def md5_handler(ctx, req):
- logger.info("md5_handler - RX " + req.encode("hex"))
+ logger.info("md5_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
def test_eap_proto_otp(dev, apdev):
"""EAP-OTP protocol tests"""
def otp_handler(ctx, req):
- logger.info("otp_handler - RX " + req.encode("hex"))
+ logger.info("otp_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
def test_eap_proto_otp_errors(dev, apdev):
"""EAP-OTP local error cases"""
def otp_handler2(ctx, req):
- logger.info("otp_handler2 - RX " + req.encode("hex"))
+ logger.info("otp_handler2 - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
def test_eap_proto_gpsk(dev, apdev):
"""EAP-GPSK protocol tests"""
def gpsk_handler(ctx, req):
- logger.info("gpsk_handler - RX " + req.encode("hex"))
+ logger.info("gpsk_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
def test_eap_proto_eke(dev, apdev):
"""EAP-EKE protocol tests"""
def eke_handler(ctx, req):
- logger.info("eke_handler - RX " + req.encode("hex"))
+ logger.info("eke_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
0xf0, 0xac, 0xcf, 0xc4, 0x66, 0xcd, 0x2d, 0xbf)
def pax_handler(ctx, req):
- logger.info("pax_handler - RX " + req.encode("hex"))
+ logger.info("pax_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
def test_eap_proto_psk(dev, apdev):
"""EAP-PSK protocol tests"""
def psk_handler(ctx, req):
- logger.info("psk_handler - RX " + req.encode("hex"))
+ logger.info("psk_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
def test_eap_proto_aka(dev, apdev):
"""EAP-AKA protocol tests"""
def aka_handler(ctx, req):
- logger.info("aka_handler - RX " + req.encode("hex"))
+ logger.info("aka_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
def test_eap_proto_aka_prime(dev, apdev):
"""EAP-AKA' protocol tests"""
def aka_prime_handler(ctx, req):
- logger.info("aka_prime_handler - RX " + req.encode("hex"))
+ logger.info("aka_prime_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
def test_eap_proto_sim(dev, apdev):
"""EAP-SIM protocol tests"""
def sim_handler(ctx, req):
- logger.info("sim_handler - RX " + req.encode("hex"))
+ logger.info("sim_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
eap_proto_ikev2_test_done = False
def ikev2_handler(ctx, req):
- logger.info("ikev2_handler - RX " + req.encode("hex"))
+ logger.info("ikev2_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
0, 0, 0, 0,
0, 0x20, 34, 0x08, 0, 28)
- def build_ike(id, next=0, exch_type=34, flags=0x00, ike=''):
+ def build_ike(id, next=0, exch_type=34, flags=0x00, ike=b''):
return struct.pack(">BBHBB2L2LBBBBLL", EAP_CODE_REQUEST, id,
4 + 1 + 1 + 28 + len(ike),
EAP_TYPE_IKEV2, flags,
logger.info("Test: Mismatch in DH Group in SAi1")
ike = build_sa(next=34)
ike += struct.pack(">BBHHH", 0, 0, 4 + 4 + 96, 12345, 0)
- ike += 96*'\x00'
+ ike += 96*b'\x00'
return build_ike(ctx['id'], next=33, ike=ike)
idx += 1
if ctx['num'] == idx:
logger.info("Test: Invalid DH public value length in SAi1")
ike = build_sa(next=34)
ike += struct.pack(">BBHHH", 0, 0, 4 + 4 + 96, 5, 0)
- ike += 96*'\x00'
+ ike += 96*b'\x00'
return build_ike(ctx['id'], next=33, ike=ike)
def build_ke(next=0):
ke = struct.pack(">BBHHH", next, 0, 4 + 4 + 192, 5, 0)
- ke += 191*'\x00'+'\x02'
+ ke += 191*b'\x00'+b'\x02'
return ke
idx += 1
logger.info("Test: Too long Ni in SAi1")
ike = build_sa(next=34)
ike += build_ke(next=40)
- ike += struct.pack(">BBH", 0, 0, 4 + 257) + 257*'\x00'
+ ike += struct.pack(">BBH", 0, 0, 4 + 257) + 257*b'\x00'
return build_ike(ctx['id'], next=33, ike=ike)
def build_ni(next=0):
- return struct.pack(">BBH", next, 0, 4 + 256) + 256*'\x00'
+ return struct.pack(">BBH", next, 0, 4 + 256) + 256*b'\x00'
def build_sai1(id):
ike = build_sa(next=34)
idx += 1
if ctx['num'] == idx:
logger.info("Test: No integrity checksum")
- ike = ''
+ ike = b''
return build_ike(ctx['id'], next=37, ike=ike)
idx += 1
idx += 1
if ctx['num'] == idx:
logger.info("Test: Invalid integrity checksum")
- ike = ''
+ ike = b''
return build_ike(ctx['id'], next=37, flags=0x20, ike=ike)
idx += 1
data = password_hash_hash + nt_response + magic1
digest = hashlib.sha1(data).digest()
- challenge = ChallengeHash(peer_challenge, auth_challenge, username)
+ challenge = ChallengeHash(peer_challenge, auth_challenge, username.encode())
data = digest + challenge + magic2
resp = hashlib.sha1(data).digest()
check_eap_capa(dev[0], "MSCHAPV2")
def mschapv2_handler(ctx, req):
- logger.info("mschapv2_handler - RX " + req.encode("hex"))
+ logger.info("mschapv2_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
idx += 1
if ctx['num'] == idx:
logger.info("Test: Failure before challenge - invalid failure challenge len")
- payload = 'C=12'
+ payload = b'C=12'
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
idx += 1
if ctx['num'] == idx:
logger.info("Test: Failure before challenge - invalid failure challenge len")
- payload = 'C=12 V=3'
+ payload = b'C=12 V=3'
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
idx += 1
if ctx['num'] == idx:
logger.info("Test: Failure before challenge - invalid failure challenge")
- payload = 'C=00112233445566778899aabbccddeefQ '
+ payload = b'C=00112233445566778899aabbccddeefQ '
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
idx += 1
if ctx['num'] == idx:
logger.info("Test: Failure before challenge - password expired")
- payload = 'E=648 R=1 C=00112233445566778899aabbccddeeff V=3 M=Password expired'
+ payload = b'E=648 R=1 C=00112233445566778899aabbccddeeff V=3 M=Password expired'
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
idx += 1
if ctx['num'] == idx:
logger.info("Test: Success after password change")
- payload = "S=1122334455667788990011223344556677889900"
+ payload = b"S=1122334455667788990011223344556677889900"
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
return struct.pack(">BBHBBBHB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + 1 + 16 + 6,
EAP_TYPE_MSCHAPV2,
- 1, 0, 4 + 1 + 16 + 6, 16) + 16*'A' + 'foobar'
+ 1, 0, 4 + 1 + 16 + 6, 16) + 16*b'A' + b'foobar'
idx += 1
if ctx['num'] == idx:
logger.info("Test: Failure - password expired")
- payload = 'E=648 R=1 C=00112233445566778899aabbccddeeff V=3 M=Password expired'
+ payload = b'E=648 R=1 C=00112233445566778899aabbccddeeff V=3 M=Password expired'
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
nt_response = data[0:24]
data = data[24:]
flags = data
- logger.info("enc_hash: " + enc_hash.encode("hex"))
- logger.info("peer_challenge: " + peer_challenge.encode("hex"))
- logger.info("nt_response: " + nt_response.encode("hex"))
- logger.info("flags: " + flags.encode("hex"))
+ logger.info("enc_hash: " + binascii.hexlify(enc_hash).decode())
+ logger.info("peer_challenge: " + binascii.hexlify(peer_challenge).decode())
+ logger.info("nt_response: " + binascii.hexlify(nt_response).decode())
+ logger.info("flags: " + binascii.hexlify(flags).decode())
auth_challenge = binascii.unhexlify("00112233445566778899aabbccddeeff")
- logger.info("auth_challenge: " + auth_challenge.encode("hex"))
+ logger.info("auth_challenge: " + binascii.hexlify(auth_challenge).decode())
auth_resp = GenerateAuthenticatorResponse("new-pw", nt_response,
peer_challenge,
auth_challenge, "user")
- payload = "S=" + auth_resp.encode('hex').upper()
- logger.info("Success message payload: " + payload)
+ payload = b"S=" + binascii.hexlify(auth_resp).decode().upper().encode()
+ logger.info("Success message payload: " + payload.decode())
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
idx += 1
if ctx['num'] == idx:
logger.info("Test: Failure - password expired")
- payload = 'E=648 R=1 C=00112233445566778899aabbccddeeff V=3 M=Password expired'
+ payload = b'E=648 R=1 C=00112233445566778899aabbccddeeff V=3 M=Password expired'
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
nt_response = data[0:24]
data = data[24:]
flags = data
- logger.info("enc_hash: " + enc_hash.encode("hex"))
- logger.info("peer_challenge: " + peer_challenge.encode("hex"))
- logger.info("nt_response: " + nt_response.encode("hex"))
- logger.info("flags: " + flags.encode("hex"))
+ logger.info("enc_hash: " + binascii.hexlify(enc_hash).decode())
+ logger.info("peer_challenge: " + binascii.hexlify(peer_challenge).decode())
+ logger.info("nt_response: " + binascii.hexlify(nt_response).decode())
+ logger.info("flags: " + binascii.hexlify(flags).decode())
auth_challenge = binascii.unhexlify("00112233445566778899aabbccddeeff")
- logger.info("auth_challenge: " + auth_challenge.encode("hex"))
+ logger.info("auth_challenge: " + binascii.hexlify(auth_challenge).decode())
auth_resp = GenerateAuthenticatorResponse("new-pw", nt_response,
peer_challenge,
auth_challenge, "user")
- payload = "S=" + auth_resp.encode('hex').upper()
- logger.info("Success message payload: " + payload)
+ payload = b"S=" + binascii.hexlify(auth_resp).decode().upper().encode()
+ logger.info("Success message payload: " + payload.decode())
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
return struct.pack(">BBHBBBHB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + 1 + 16 + 6,
EAP_TYPE_MSCHAPV2,
- 1, 0, 4 + 1 + 16 + 6, 16) + 16*'A' + 'foobar'
+ 1, 0, 4 + 1 + 16 + 6, 16) + 16*b'A' + b'foobar'
idx += 1
if ctx['num'] == idx:
logger.info("Test: Failure - authentication failure")
- payload = 'E=691 R=1 C=00112233445566778899aabbccddeeff V=3 M=Authentication failed'
+ payload = b'E=691 R=1 C=00112233445566778899aabbccddeeff V=3 M=Authentication failed'
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
return struct.pack(">BBHBBBHB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + 1 + 16 + 6,
EAP_TYPE_MSCHAPV2,
- 1, 0, 4 + 1 + 16 + 6, 16) + 16*'A' + 'foobar'
+ 1, 0, 4 + 1 + 16 + 6, 16) + 16*b'A' + b'foobar'
idx += 1
if ctx['num'] == idx:
logger.info("Test: Failure - authentication failure")
- payload = 'E=691 R=1 C=00112233445566778899aabbccddeeff V=3 M=Authentication failed (2)'
+ payload = b'E=691 R=1 C=00112233445566778899aabbccddeeff V=3 M=Authentication failed (2)'
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
return struct.pack(">BBHBBBHB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + 1 + 16 + 6,
EAP_TYPE_MSCHAPV2,
- 1, 0, 4 + 1 + 16 + 6 + 1, 16) + 16*'A' + 'foobar'
+ 1, 0, 4 + 1 + 16 + 6 + 1, 16) + 16*b'A' + b'foobar'
return None
def mschapv2_fail_password_expired(ctx):
logger.info("Test: Failure before challenge - password expired")
- payload = 'E=648 R=1 C=00112233445566778899aabbccddeeff V=3 M=Password expired'
+ payload = b'E=648 R=1 C=00112233445566778899aabbccddeeff V=3 M=Password expired'
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
def mschapv2_success_after_password_change(ctx, req=None):
logger.info("Test: Success after password change")
if req is None or len(req) != 591:
- payload = "S=1122334455667788990011223344556677889900"
+ payload = b"S=1122334455667788990011223344556677889900"
else:
data = req[9:]
enc_pw = data[0:516]
nt_response = data[0:24]
data = data[24:]
flags = data
- logger.info("enc_hash: " + enc_hash.encode("hex"))
- logger.info("peer_challenge: " + peer_challenge.encode("hex"))
- logger.info("nt_response: " + nt_response.encode("hex"))
- logger.info("flags: " + flags.encode("hex"))
+ logger.info("enc_hash: " + binascii.hexlify(enc_hash).decode())
+ logger.info("peer_challenge: " + binascii.hexlify(peer_challenge).decode())
+ logger.info("nt_response: " + binascii.hexlify(nt_response).decode())
+ logger.info("flags: " + binascii.hexlify(flags).decode())
auth_challenge = binascii.unhexlify("00112233445566778899aabbccddeeff")
- logger.info("auth_challenge: " + auth_challenge.encode("hex"))
+ logger.info("auth_challenge: " + binascii.hexlify(auth_challenge).decode())
auth_resp = GenerateAuthenticatorResponse("new-pw", nt_response,
peer_challenge,
auth_challenge, "user")
- payload = "S=" + auth_resp.encode('hex').upper()
+ payload = b"S=" + binascii.hexlify(auth_resp).decode().upper().encode()
return struct.pack(">BBHBBBH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + len(payload),
EAP_TYPE_MSCHAPV2,
3, 0, 4 + len(payload)) + payload
def mschapv2_handler(ctx, req):
- logger.info("mschapv2_handler - RX " + req.encode("hex"))
+ logger.info("mschapv2_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
eap_proto_pwd_test_wait = False
def pwd_handler(ctx, req):
- logger.info("pwd_handler - RX " + req.encode("hex"))
+ logger.info("pwd_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
idx += 1
if ctx['num'] == idx:
logger.info("Test: Commit payload with all zeros values --> Shared key at infinity")
- payload = struct.pack(">B", 0x02) + 96*'\0'
+ payload = struct.pack(">B", 0x02) + 96*b'\0'
return struct.pack(">BBHB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + len(payload), EAP_TYPE_PWD) + payload
idx += 1
if ctx['num'] == idx:
logger.info("Test: Confirm payload with incorrect value")
- payload = struct.pack(">B", 0x03) + 32*'\0'
+ payload = struct.pack(">B", 0x03) + 32*b'\0'
return struct.pack(">BBHB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + len(payload), EAP_TYPE_PWD) + payload
eap_proto_erp_test_done = False
def erp_handler(ctx, req):
- logger.info("erp_handler - RX " + req.encode("hex"))
+ logger.info("erp_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] += 1
idx += 1
if ctx['num'] == idx:
logger.info("Test: Zero-length TVs/TLVs")
- payload = ""
+ payload = b""
return struct.pack(">BBHBB", EAP_CODE_INITIATE, ctx['id'],
4 + 1 + 1 + len(payload),
EAP_ERP_TYPE_REAUTH_START, 0) + payload
eap_proto_expanded_test_done = False
def expanded_handler(ctx, req):
- logger.info("expanded_handler - RX " + req.encode("hex"))
+ logger.info("expanded_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] += 1
return struct.pack(">BBHBBBHB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 4 + 1 + 16 + 6,
EAP_TYPE_MSCHAPV2,
- 1, 0, 4 + 1 + 16 + 6, 16) + 16*'A' + 'foobar'
+ 1, 0, 4 + 1 + 16 + 6, 16) + 16*b'A' + b'foobar'
idx += 1
if ctx['num'] == idx:
logger.info("Test: Invalid expanded frame type")
eap_proto_tls_test_wait = False
def tls_handler(ctx, req):
- logger.info("tls_handler - RX " + req.encode("hex"))
+ logger.info("tls_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] += 1
idx += 1
if ctx['num'] == idx:
logger.info("Test: Fragmented TLS message (long; first)")
- payload = 1450*'A'
+ payload = 1450*b'A'
return struct.pack(">BBHBBL", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + 4 + len(payload),
EAP_TYPE_TLS, 0xc0, 65536) + payload
if ctx['num'] == idx:
logger.info("Test: Fragmented TLS message (long; cont %d)" % i)
eap_proto_tls_test_wait = True
- payload = 1470*'A'
+ payload = 1470*b'A'
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(payload),
EAP_TYPE_TLS, 0x40) + payload
eap_proto_tnc_test_done = False
def tnc_handler(ctx, req):
- logger.info("tnc_handler - RX " + req.encode("hex"))
+ logger.info("tnc_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] += 1
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = "FOO"
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b"FOO"
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = "</TNCCS-Batch><TNCCS-Batch>"
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b"</TNCCS-Batch><TNCCS-Batch>"
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = "<TNCCS-Batch foo=3></TNCCS-Batch>"
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b"<TNCCS-Batch foo=3></TNCCS-Batch>"
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = "<TNCCS-Batch BatchId=123456789></TNCCS-Batch>"
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b"<TNCCS-Batch BatchId=123456789></TNCCS-Batch>"
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = "<TNCCS-Batch BatchId=2><IMC-IMV-Message><TNCC-TNCS-Message></TNCCS-Batch>"
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b"<TNCCS-Batch BatchId=2><IMC-IMV-Message><TNCC-TNCS-Message></TNCCS-Batch>"
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = "<TNCCS-Batch BatchId=2><IMC-IMV-Message></IMC-IMV-Message><TNCC-TNCS-Message></TNCC-TNCS-Message></TNCCS-Batch>"
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b"<TNCCS-Batch BatchId=2><IMC-IMV-Message></IMC-IMV-Message><TNCC-TNCS-Message></TNCC-TNCS-Message></TNCCS-Batch>"
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = "<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><XML></TNCC-TNCS-Message></TNCCS-Batch>"
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b"<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><XML></TNCC-TNCS-Message></TNCCS-Batch>"
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = "<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type></TNCC-TNCS-Message></TNCCS-Batch>"
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b"<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type></TNCC-TNCS-Message></TNCCS-Batch>"
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = "<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><Base64>abc</TNCC-TNCS-Message></TNCCS-Batch>"
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b"<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><Base64>abc</TNCC-TNCS-Message></TNCCS-Batch>"
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = "<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><Base64>aGVsbG8=</Base64></TNCC-TNCS-Message></TNCCS-Batch>"
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b"<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><Base64>aGVsbG8=</Base64></TNCC-TNCS-Message></TNCCS-Batch>"
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = "<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><XML>hello</XML></TNCC-TNCS-Message></TNCCS-Batch>"
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b"<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><XML>hello</XML></TNCC-TNCS-Message></TNCCS-Batch>"
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = '<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><XML><TNCCS-Recommendation foo=1></TNCCS-Recommendation></XML></TNCC-TNCS-Message></TNCCS-Batch>'
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b'<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><XML><TNCCS-Recommendation foo=1></TNCCS-Recommendation></XML></TNCC-TNCS-Message></TNCCS-Batch>'
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = '<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><XML><TNCCS-Recommendation type="none"></TNCCS-Recommendation></XML></TNCC-TNCS-Message></TNCCS-Batch>'
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b'<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><XML><TNCCS-Recommendation type="none"></TNCCS-Recommendation></XML></TNCC-TNCS-Message></TNCCS-Batch>'
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
EAP_TYPE_TNC, 0x21)
idx += 1
if ctx['num'] == idx:
- logger.info("Received TNCCS-Batch: " + req[6:])
- resp = '<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><XML><TNCCS-Recommendation type="isolate"></TNCCS-Recommendation></XML></TNCC-TNCS-Message></TNCCS-Batch>'
+ logger.info("Received TNCCS-Batch: " + binascii.hexlify(req[6:]).decode())
+ resp = b'<TNCCS-Batch BatchId=2><TNCC-TNCS-Message><Type>00000001</Type><XML><TNCCS-Recommendation type="isolate"></TNCCS-Recommendation></XML></TNCC-TNCS-Message></TNCCS-Batch>'
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(resp),
EAP_TYPE_TNC, 0x01) + resp
"""EAP protocol tests for canned EAP-Success after identity"""
check_eap_capa(dev[0], "MD5")
def eap_canned_success_handler(ctx, req):
- logger.info("eap_canned_success_handler - RX " + req.encode("hex"))
+ logger.info("eap_canned_success_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
eap_proto_wsc_test_done = False
def wsc_handler(ctx, req):
- logger.info("wsc_handler - RX " + req.encode("hex"))
+ logger.info("wsc_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] += 1
logger.info("Send EAP-FAST/Start")
return struct.pack(">BBHBBHH", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + 4 + 16,
- EAP_TYPE_FAST, 0x21, 4, 16) + 16*'A'
+ EAP_TYPE_FAST, 0x21, 4, 16) + 16*b'A'
def test_eap_fast_proto(dev, apdev):
"""EAP-FAST Phase protocol testing"""
eap_fast_proto_ctx = None
def eap_handler(ctx, req):
- logger.info("eap_handler - RX " + req.encode("hex"))
+ logger.info("eap_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
idx += 1
if ctx['num'] == idx:
logger.info("EAP-FAST: TLS processing failed")
- data = 'ABCDEFGHIK'
+ data = b'ABCDEFGHIK'
return struct.pack(">BBHBB", EAP_CODE_REQUEST, ctx['id'],
4 + 1 + 1 + len(data),
EAP_TYPE_FAST, 0x01) + data
EAP_TYPE_FAST, 0x01) + data
def eap_handler(ctx, req):
- logger.info("eap_handler - RX " + req.encode("hex"))
+ logger.info("eap_handler - RX " + binascii.hexlify(req).decode())
if 'num' not in ctx:
ctx['num'] = 0
ctx['num'] = ctx['num'] + 1
dict = pyrad.dictionary.Dictionary("dictionary.radius")
srv = pyrad.client.Client(server="127.0.0.1", acctport=3799,
- secret="secret", dict=dict)
+ secret=b"secret", dict=dict)
srv.retries = 1
srv.timeout = 1
logger.info("Disconnect-Request with incorrect secret")
- req = radius_das.DisconnectPacket(dict=dict, secret="incorrect",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"incorrect",
User_Name="foo",
NAS_Identifier="localhost",
Event_Timestamp=int(time.time()))
logger.info("Disconnect-Request with incorrect secret properly ignored")
logger.info("Disconnect-Request without Event-Timestamp")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
User_Name="psk.user@example.com")
logger.debug(req)
try:
logger.info("Disconnect-Request without Event-Timestamp properly ignored")
logger.info("Disconnect-Request with non-matching Event-Timestamp")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
User_Name="psk.user@example.com",
Event_Timestamp=123456789)
logger.debug(req)
logger.info("Disconnect-Request with non-matching Event-Timestamp properly ignored")
logger.info("Disconnect-Request with unsupported attribute")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
User_Name="foo",
User_Password="foo",
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 401)
logger.info("Disconnect-Request with invalid Calling-Station-Id")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
User_Name="foo",
Calling_Station_Id="foo",
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 407)
logger.info("Disconnect-Request with mismatching User-Name")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
User_Name="foo",
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503)
logger.info("Disconnect-Request with mismatching Calling-Station-Id")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
Calling_Station_Id="12:34:56:78:90:aa",
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503)
logger.info("Disconnect-Request with mismatching Acct-Session-Id")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
Acct_Session_Id="12345678-87654321",
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503)
logger.info("Disconnect-Request with mismatching Acct-Session-Id (len)")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
Acct_Session_Id="12345678",
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503)
logger.info("Disconnect-Request with mismatching Acct-Multi-Session-Id")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
Acct_Multi_Session_Id="12345678+87654321",
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503)
logger.info("Disconnect-Request with mismatching Acct-Multi-Session-Id (len)")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
Acct_Multi_Session_Id="12345678",
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503)
logger.info("Disconnect-Request with no session identification attributes")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503)
raise Exception("Unexpected disconnection")
logger.info("Disconnect-Request with mismatching NAS-IP-Address")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_IP_Address="192.168.3.4",
Acct_Session_Id=id,
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 403)
logger.info("Disconnect-Request with mismatching NAS-Identifier")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_Identifier="unknown.example.com",
Acct_Session_Id=id,
Event_Timestamp=int(time.time()))
raise Exception("Unexpected disconnection")
logger.info("Disconnect-Request with matching Acct-Session-Id")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_IP_Address="127.0.0.1",
NAS_Identifier="nas.example.com",
Acct_Session_Id=id,
logger.info("Disconnect-Request with matching Acct-Multi-Session-Id")
sta = hapd.get_sta(addr)
multi_sess_id = sta['authMultiSessionId']
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_IP_Address="127.0.0.1",
NAS_Identifier="nas.example.com",
Acct_Multi_Session_Id=multi_sess_id,
dev[0].wait_connected(timeout=10, error="Re-connection timed out")
logger.info("Disconnect-Request with matching User-Name")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_Identifier="nas.example.com",
User_Name="psk.user@example.com",
Event_Timestamp=int(time.time()))
dev[0].wait_connected(timeout=10, error="Re-connection timed out")
logger.info("Disconnect-Request with matching Calling-Station-Id")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_IP_Address="127.0.0.1",
Calling_Station_Id=addr,
Event_Timestamp=int(time.time()))
dev[0].wait_connected(timeout=10, error="Re-connection timed out")
logger.info("Disconnect-Request with matching Calling-Station-Id and non-matching CUI")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
Calling_Station_Id=addr,
Chargeable_User_Identity="foo@example.com",
Event_Timestamp=int(time.time()))
eap="GPSK", identity="gpsk-cui",
password="abcdefghijklmnop0123456789abcdef",
scan_freq="2412")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
Chargeable_User_Identity="gpsk-chargeable-user-identity",
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectACK)
connect(dev[2], "radius-das")
logger.info("Disconnect-Request with matching User-Name - multiple sessions matching")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_Identifier="nas.example.com",
User_Name="psk.user@example.com",
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, error_cause=508)
logger.info("Disconnect-Request with User-Name matching multiple sessions, Calling-Station-Id only one")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_Identifier="nas.example.com",
Calling_Station_Id=addr,
User_Name="psk.user@example.com",
multi_sess_id = sta['authMultiSessionId']
dev[0].request("DISCONNECT")
dev[0].wait_disconnected(timeout=10)
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_IP_Address="127.0.0.1",
NAS_Identifier="nas.example.com",
Acct_Multi_Session_Id=multi_sess_id,
dev[0].wait_disconnected(timeout=10)
dev[2].request("DISCONNECT")
dev[2].wait_disconnected(timeout=10)
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_IP_Address="127.0.0.1",
NAS_Identifier="nas.example.com",
User_Name="psk.user@example.com",
logger.info("Disconnect-Request with matching CUI after disassociation")
dev[1].request("DISCONNECT")
dev[1].wait_disconnected(timeout=10)
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_IP_Address="127.0.0.1",
NAS_Identifier="nas.example.com",
Chargeable_User_Identity="gpsk-chargeable-user-identity",
dev[0].wait_connected(timeout=15)
dev[0].request("DISCONNECT")
dev[0].wait_disconnected(timeout=10)
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_IP_Address="127.0.0.1",
NAS_Identifier="nas.example.com",
Calling_Station_Id=addr,
send_and_check_reply(srv, req, pyrad.packet.DisconnectACK)
logger.info("Disconnect-Request with mismatching Calling-Station-Id after disassociation")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_IP_Address="127.0.0.1",
NAS_Identifier="nas.example.com",
Calling_Station_Id=addr,
hmac_obj.update(struct.pack("B", req.id))
# request attributes
- req.AddAttribute("Message-Authenticator", 16*"\x00")
+ req.AddAttribute("Message-Authenticator", 16*b"\x00")
attrs = req._PktEncodeAttributes()
# Length
flen = 4 + 16 + len(attrs)
hmac_obj.update(struct.pack(">H", flen))
- hmac_obj.update(16*"\x00") # all zeros Authenticator in calculation
+ hmac_obj.update(16*b"\x00") # all zeros Authenticator in calculation
hmac_obj.update(attrs)
del req[80]
req.AddAttribute("Message-Authenticator", hmac_obj.digest())
dict = pyrad.dictionary.Dictionary("dictionary.radius")
srv = pyrad.client.Client(server="127.0.0.1", acctport=3799,
- secret="secret", dict=dict)
+ secret=b"secret", dict=dict)
srv.retries = 1
srv.timeout = 1
logger.info("Disconnect-Request with unsupported attribute")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_IP_Address="127.0.0.1",
NAS_Identifier="nas.example.com",
Calling_Station_Id=addr,
logger.info("Disconnect-Request with non-matching Event-Timestamp properly ignored")
logger.info("Disconnect-Request with unsupported attribute")
- req = radius_das.DisconnectPacket(dict=dict, secret="secret",
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
NAS_IP_Address="127.0.0.1",
NAS_Identifier="nas.example.com",
Calling_Station_Id=addr,
dict = pyrad.dictionary.Dictionary("dictionary.radius")
srv = pyrad.client.Client(server="127.0.0.1", acctport=3799,
- secret="secret", dict=dict)
+ secret=b"secret", dict=dict)
srv.retries = 1
srv.timeout = 1
# hostapd does not currently support CoA-Request, so NAK is expected
logger.info("CoA-Request with matching Acct-Session-Id")
- req = radius_das.CoAPacket(dict=dict, secret="secret",
+ req = radius_das.CoAPacket(dict=dict, secret=b"secret",
Acct_Session_Id=id,
Event_Timestamp=int(time.time()))
send_and_check_reply(srv, req, pyrad.packet.CoANAK, error_cause=405)
logger.info("Add Message-Authenticator")
if self.t_events['wrong_secret'].is_set():
logger.info("Use incorrect RADIUS shared secret")
- pw = "incorrect"
+ pw = b"incorrect"
else:
pw = reply.secret
hmac_obj = hmac.new(pw)
hmac_obj.update(struct.pack("B", reply.id))
# reply attributes
- reply.AddAttribute("Message-Authenticator",
- "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00")
+ reply.AddAttribute("Message-Authenticator", 16*b"\x00")
attrs = reply._PktEncodeAttributes()
# Length
srv = TestServer(dict=pyrad.dictionary.Dictionary("dictionary.radius"),
authport=18138, acctport=18139)
srv.hosts["127.0.0.1"] = pyrad.server.RemoteHost("127.0.0.1",
- "radius",
+ b"radius",
"localhost")
srv.BindToAddress("")
t_events = {}
t.join()
def build_tunnel_password(secret, authenticator, psk):
- a = "\xab\xcd"
+ a = b"\xab\xcd"
+ psk = psk.encode()
padlen = 16 - (1 + len(psk)) % 16
if padlen == 16:
padlen = 0
cc = bytearray(pp[i] ^ bb[i] for i in range(len(bb)))
cc_all += cc
b = hashlib.md5(secret + cc).digest()
- data = '\x00' + a + bytes(cc_all)
+ data = b'\x00' + a + bytes(cc_all)
return data
def start_radius_psk_server(psk, invalid_code=False, acct_interim_interval=0,
srv = TestServer(dict=pyrad.dictionary.Dictionary("dictionary.radius"),
authport=18138, acctport=18139)
srv.hosts["127.0.0.1"] = pyrad.server.RemoteHost("127.0.0.1",
- "radius",
+ b"radius",
"localhost")
srv.BindToAddress("")
t_events = {}
hmac_obj.update(struct.pack("B", req.id))
# request attributes
- req.AddAttribute("Message-Authenticator",
- "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00")
+ req.AddAttribute("Message-Authenticator", 16*b"\x00")
attrs = req._PktEncodeAttributes()
# Length
dict = pyrad.dictionary.Dictionary("dictionary.radius")
client = pyrad.client.Client(server="127.0.0.1", authport=1812,
- secret="radius", dict=dict)
+ secret=b"radius", dict=dict)
client.retries = 1
client.timeout = 1
# unexpected State
req = client.CreateAuthPacket(code=pyrad.packet.AccessRequest,
User_Name="foo")
- req['State'] = 'foo-state'
+ req['State'] = b'foo-state'
add_message_auth(req)
reply = client.SendPacket(req)
if reply.code != pyrad.packet.AccessReject:
srv = TestServer(dict=pyrad.dictionary.Dictionary("dictionary.radius"),
authport=18138, acctport=18139)
srv.hosts["127.0.0.1"] = pyrad.server.RemoteHost("127.0.0.1",
- "radius",
+ b"radius",
"localhost")
srv.BindToAddress("")
t_events = {}