sindoring
8c8e982fc3
Keep launcher tickets mandatory by default and compile password authentication only into the dedicated local image. Build the shared local image once and add an end-to-end SRP login check.
137 lines
6.4 KiB
Python
137 lines
6.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise local SRP login without launcher tickets using a disposable account."""
|
|
|
|
import argparse
|
|
import hashlib
|
|
from pathlib import Path
|
|
import secrets
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
|
|
|
|
COMPOSE = Path(__file__).resolve().with_name("compose.sh")
|
|
N = int("894B645E89E1535BBDAD5B8B290650530801B18EBFBF5E8FAB3C82872A3E9BB7", 16)
|
|
|
|
|
|
def sha(data):
|
|
return hashlib.sha1(data).digest()
|
|
|
|
|
|
def number_bytes(value, size=None):
|
|
return value.to_bytes(size if size is not None else (value.bit_length() + 7) // 8, "little")
|
|
|
|
|
|
def sql(statement):
|
|
result = subprocess.run(
|
|
[str(COMPOSE), "exec", "-T", "db", "sh", "-c",
|
|
'MYSQL_PWD="$(cat /run/secrets/db_password)" exec mariadb -uvmangos --batch --skip-column-names realmd'],
|
|
input=statement, text=True, capture_output=True, check=True,
|
|
)
|
|
return result.stdout.strip()
|
|
|
|
|
|
def receive(connection, size):
|
|
result = b""
|
|
while len(result) < size:
|
|
chunk = connection.recv(size - len(result))
|
|
if not chunk:
|
|
raise AssertionError("Authentication connection closed unexpectedly")
|
|
result += chunk
|
|
return result
|
|
|
|
|
|
def challenge(connection, username):
|
|
name = username.encode("ascii")
|
|
body = struct.pack("<4sBBBH4s4s4sIIB", b"WoW\0", 1, 12, 1, 5875,
|
|
b"68x\0", b"niW\0", b"URur", 0, 0x0100007F, len(name)) + name
|
|
connection.sendall(struct.pack("<BBH", 0, 0, len(body)) + body)
|
|
prefix = receive(connection, 3)
|
|
assert prefix[:2] == b"\0\0", prefix
|
|
if prefix[2]:
|
|
return prefix[2], None
|
|
public = int.from_bytes(receive(connection, 32), "little")
|
|
generator = int.from_bytes(receive(connection, receive(connection, 1)[0]), "little")
|
|
modulus = int.from_bytes(receive(connection, receive(connection, 1)[0]), "little")
|
|
salt = int.from_bytes(receive(connection, 32), "little")
|
|
receive(connection, 16) # Version challenge seed.
|
|
assert receive(connection, 1) == b"\0", "Disposable account unexpectedly requires a PIN"
|
|
assert modulus == N and generator == 7
|
|
return 0, (public, generator, salt)
|
|
|
|
|
|
def login(port, username, password, integrity, expect_success):
|
|
with socket.create_connection(("127.0.0.1", port), timeout=10) as connection:
|
|
code, parameters = challenge(connection, username)
|
|
assert code == 0, f"Logon challenge rejected with code {code}"
|
|
public, generator, salt = parameters
|
|
private = secrets.randbits(152)
|
|
client_public = pow(generator, private, N)
|
|
public_bytes = number_bytes(client_public)
|
|
server_bytes = number_bytes(public)
|
|
salt_bytes = number_bytes(salt)
|
|
scramble = int.from_bytes(sha(public_bytes + server_bytes), "little")
|
|
exponent = int.from_bytes(sha(salt_bytes + sha(f"{username}:{password}".encode())), "little")
|
|
shared = pow((public - 3 * pow(generator, exponent, N)) % N, private + scramble * exponent, N)
|
|
shared_bytes = number_bytes(shared, 32)
|
|
even, odd = sha(shared_bytes[::2]), sha(shared_bytes[1::2])
|
|
key = bytes(value for pair in zip(even, odd) for value in pair)
|
|
key_bytes = number_bytes(int.from_bytes(key, "little"))
|
|
xor_hash = bytes(a ^ b for a, b in zip(sha(number_bytes(N)), sha(number_bytes(generator))))
|
|
proof = sha(number_bytes(int.from_bytes(xor_hash, "little")) + sha(username.encode())
|
|
+ salt_bytes + public_bytes + server_bytes + key_bytes)
|
|
wire_public = number_bytes(client_public, 32)
|
|
version_proof = sha(wire_public + integrity)
|
|
connection.sendall(b"\x01" + wire_public + proof + version_proof + b"\0\0")
|
|
result = receive(connection, 2)
|
|
assert result[0] == 1
|
|
if not expect_success:
|
|
assert result[1] == 4, f"Wrong password was not rejected: {result.hex()}"
|
|
return
|
|
assert result[1] == 0, f"Logon proof rejected with code {result[1]}"
|
|
expected_proof = sha(public_bytes + number_bytes(int.from_bytes(proof, "little")) + key_bytes)
|
|
assert receive(connection, 20) == expected_proof, "Server SRP proof mismatch"
|
|
receive(connection, 4) # Vanilla survey id.
|
|
stored = sql(f"SELECT sessionkey FROM account WHERE username='{username}';")
|
|
assert int(stored, 16) == int.from_bytes(key, "little"), "Session key was not published"
|
|
connection.sendall(b"\x10\0\0\0\0")
|
|
assert receive(connection, 1) == b"\x10"
|
|
size = int.from_bytes(receive(connection, 2), "little")
|
|
realms = receive(connection, size)
|
|
assert b"127.0.0.1:8085\0" in realms, "Local realm missing from authenticated realm list"
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--port", type=int, default=3724)
|
|
parser.add_argument("--required-port", type=int, help="Optional local ticket-required realmd for the negative test")
|
|
args = parser.parse_args()
|
|
username = "LCLSMOKE" + secrets.token_hex(3).upper()
|
|
password = secrets.token_hex(8).upper()
|
|
salt = secrets.randbits(255) | (1 << 255)
|
|
exponent = int.from_bytes(sha(number_bytes(salt) + sha(f"{username}:{password}".encode())), "little")
|
|
verifier = pow(7, exponent, N)
|
|
sql(f"INSERT INTO account (username,v,s,token_key,email_verif) "
|
|
f"VALUES ('{username}','{verifier:X}','{salt:X}','',1);")
|
|
try:
|
|
assert sql(f"SELECT COUNT(*) FROM launcher_ticket t JOIN account a ON a.id=t.account_id "
|
|
f"WHERE a.username='{username}';") == "0"
|
|
integrity = bytes.fromhex(sql("SELECT integrity_hash FROM allowed_clients "
|
|
"WHERE build=5875 AND platform='x86' AND os='Win' LIMIT 1;"))
|
|
if args.required_port:
|
|
with socket.create_connection(("127.0.0.1", args.required_port), timeout=10) as connection:
|
|
code, _ = challenge(connection, username)
|
|
assert code == 4, f"Ticket-required build accepted a ticketless account: {code}"
|
|
print("PASS: ticket-required build rejects an account without a launcher ticket")
|
|
login(args.port, username, password + "WRONG", integrity, False)
|
|
print("PASS: local build rejects an incorrect password")
|
|
login(args.port, username, password, integrity, True)
|
|
print("PASS: ticketless SRP login, server proof, session key and local realm list")
|
|
finally:
|
|
sql(f"DELETE FROM account WHERE username='{username}';")
|
|
print("Disposable account removed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|