barneygale / asyncvnc Goto Github PK

import asyncio
import asyncvnc

from os import urandom
from typing import Optional

from asyncvnc import Client, Clipboard, Keyboard, Mouse, Video, read_int
from cryptography.hazmat.primitives.asymmetric import padding, rsa
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.serialization import load_der_public_key
from unicrypto.symmetric import DES, MODE_ECB

def pack_ard(data):
    data = data.encode("utf-8") + b"\x00"
    if len(data) < 64:
        data += urandom(64 - len(data))
    else:
        data = data[:64]
    return data

@classmethod
async def create(
    cls,
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter,
    username: Optional[str] = None,
    password: Optional[str] = None,
    host_key: Optional[rsa.RSAPublicKey] = None,
) -> Client:

    intro = await reader.readline()
    if intro[:4] != b"RFB ":
        raise ValueError("not a VNC server")
    writer.write(b"RFB 003.008\n")

    auth_types = set(await reader.readexactly(await read_int(reader, 1)))
    for auth_type in (33, 1, 2):
        if auth_type in auth_types:
            writer.write(auth_type.to_bytes(1, "big"))
            break
    else:
        raise ValueError(f"unsupported auth types: {auth_types}")

    # Apple authentication
    if auth_type == 33:
        if username is None or password is None:
            raise ValueError("server requires username and password")
        if host_key is None:
            writer.write(b"\x00\x00\x00\x0a\x01\x00RSA1\x00\x00\x00\x00")
            await reader.readexactly(4)  # packet length
            await reader.readexactly(2)  # packet version
            host_key_length = await read_int(reader, 4)
            host_key = await reader.readexactly(host_key_length)
            host_key = load_der_public_key(host_key)
            await reader.readexactly(1)  # unknown
        aes_key = urandom(16)
        cipher = Cipher(algorithms.AES(aes_key), modes.ECB())
        encryptor = cipher.encryptor()
        credentials = pack_ard(username) + pack_ard(password)
        writer.write(
            b"\x00\x00\x01\x8a\x01\x00RSA1"
            + b"\x00\x01"
            + encryptor.update(credentials)
            + b"\x00\x01"
            + host_key.encrypt(aes_key, padding=padding.PKCS1v15())
        )
        await reader.readexactly(4)  # unknown

    # VNC authentication
    if auth_type == 2:
        if password is None:
            raise ValueError("server requires password")
        des_key = password.encode("ascii")[:8].ljust(8, b"\x00")
        des_key = bytes(int(bin(n)[:1:-1], 2) for n in des_key)
        encryptor = Cipher(algorithms.TripleDES(des_key), modes.ECB()).encryptor()
        challenge = await reader.readexactly(16)

        password = password.ljust(8, "\x00").encode("ascii")
        password = password[:8]
        # converting password to key
        newkey = b""
        for ki in range(len(password)):
            bsrc = password[ki]
            btgt = 0
            for i in range(8):
                if bsrc & (1 << i):
                    btgt = btgt | (1 << 7 - i)
            newkey += bytes([btgt])
        ctx = DES(newkey, mode=MODE_ECB, IV=None)

        _aardwolf_response = ctx.encrypt(challenge)
        _asyncvnc_response = encryptor.update(challenge) + encryptor.finalize()

        print("aardwolf: ", _aardwolf_response, "\nasyncvnc: ", _asyncvnc_response)

        writer.write(_aardwolf_response)

    auth_result = await read_int(reader, 4)
    if auth_result == 0:
        return cls(
            reader=reader,
            writer=writer,
            host_key=host_key,
            clipboard=Clipboard(writer),
            keyboard=Keyboard(writer),
            mouse=Mouse(writer),
            video=await Video.create(reader, writer),
        )
    elif auth_result == 1:
        raise PermissionError("Auth failed")
    elif auth_result == 2:
        raise PermissionError("Auth failed (too many attempts)")
    else:
        reason = await reader.readexactly(auth_result)
        raise PermissionError(reason.decode("utf-8"))

Client.create = create

async def run_client():
    async with asyncvnc.connect("10.9.0.9", port=5900, password="epsa"):
        pass

asyncio.run(run_client())