yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

moonlight-sunshine-accept: use pathlib and fix types

Browse Source
This commit is contained in:
Jörg Thalheim
2024-09-02 17:32:17 +02:00
parent 7fc923f310
commit 71e026dd62
11 changed files with 92 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
pkgs/moonlight-sunshine-accept/moonlight_sunshine_accept/moonlight/init_certificates.py
View File

@@ -1,6 +1,5 @@
import argparse
import datetime
import os
from datetime import timedelta
from pathlib import Path
@@ -54,7 +53,7 @@ def private_key_to_pem(private_key: rsa.RSAPrivateKey) -> bytes:
return pem_private_key
def init_credentials() -> (str, str):
def init_credentials() -> tuple[bytes, bytes]:
private_key = generate_private_key()
certificate = generate_certificate(private_key)
private_key_pem = private_key_to_pem(private_key)
@@ -63,15 +62,11 @@ def init_credentials() -> (str, str):
def write_credentials(_args: argparse.Namespace) -> None:
pem_certificate, pem_private_key = init_credentials()
credentials_path = os.getcwd() + "credentials"
credentials_path = Path.cwd() / "credentials"
Path(credentials_path).mkdir(parents=True, exist_ok=True)
cacaert_path = os.path.join(credentials_path, "cacert.pem")
with open(cacaert_path, mode="wb") as file:
file.write(pem_certificate)
cakey_path = os.path.join(credentials_path, "cakey.pem")
with open(cakey_path, mode="wb") as file:
file.write(pem_private_key)
(credentials_path / "cacert.pem").write_bytes(pem_certificate)
(credentials_path / "cakey.pem").write_bytes(pem_private_key)
print("Finished writing moonlight credentials")

16
pkgs/moonlight-sunshine-accept/moonlight_sunshine_accept/moonlight/join.py
View File

@@ -33,8 +33,7 @@ def send_join_request_api(host: str, port: int) -> bool:
with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
s.connect((host, port))
json_body = {"type": "api", "pin": pin}
json_body = json.dumps(json_body)
json_body = json.dumps({"type": "api", "pin": pin})
request = (
f"POST / HTTP/1.1\r\n"
f"Content-Type: application/json\r\n"
@@ -62,14 +61,15 @@ def send_join_request_native(host: str, port: int, cert: str) -> bool:
with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
s.connect((host, port))
encoded_cert = base64.urlsafe_b64encode(cert.encode("utf-8")).decode("utf-8")
json_body = {"type": "native", "uuid": uuid, "cert": encoded_cert}
json_body = json.dumps(json_body)
json_body_str = json.dumps(
{"type": "native", "uuid": uuid, "cert": encoded_cert}
)
request = (
f"POST / HTTP/1.1\r\n"
f"Content-Type: application/json\r\n"
f"Content-Length: {len(json_body)}\r\n"
f"Content-Length: {len(json_body_str)}\r\n"
f"Connection: close\r\n\r\n"
f"{json_body}"
f"{json_body_str}"
)
try:
s.sendall(request.encode("utf-8"))
@@ -78,7 +78,7 @@ def send_join_request_native(host: str, port: int, cert: str) -> bool:
lines = response.split("\n")
body = "\n".join(lines[2:])[2:]
print(body)
return body
return True
except Exception as e:
print(f"An error occurred: {e}")
# TODO: fix
@@ -98,6 +98,7 @@ def send_join_request_native(host: str, port: int, cert: str) -> bool:
print(f"Failed to decode JSON: {e}")
pos = e.pos
print(f"Failed to decode JSON: unexpected character {response[pos]}")
return False
def join(args: argparse.Namespace) -> None:
@@ -113,6 +114,7 @@ def join(args: argparse.Namespace) -> None:
# TODO: If cert is not provided parse from config
# cert = args.cert
cert = get_moonlight_certificate()
assert port is not None
if send_join_request(host, port, cert):
print(f"Successfully joined sunshine host: {host}")
else:

6
pkgs/moonlight-sunshine-accept/moonlight_sunshine_accept/moonlight/run.py
View File

@@ -4,8 +4,8 @@ import threading
class MoonlightPairing:
def __init__(self) -> "MoonlightPairing":
self.process = None
def __init__(self) -> None:
self.process: subprocess.Popen | None = None
self.output = ""
self.found = threading.Event()
@@ -45,6 +45,8 @@ class MoonlightPairing:
self.process.wait()
def stream_output(self, target_string: str) -> None:
assert self.process is not None
assert self.process.stdout is not None
for line in iter(self.process.stdout.readline, b""):
line = line.decode()
self.output += line

41
pkgs/moonlight-sunshine-accept/moonlight_sunshine_accept/moonlight/state.py
View File

@@ -3,21 +3,20 @@ import os
import random
import string
from configparser import ConfigParser, DuplicateSectionError, NoOptionError
from pathlib import Path
def moonlight_config_dir() -> str:
return os.path.join(
os.path.expanduser("~"), ".config", "Moonlight Game Streaming Project"
)
def moonlight_config_dir() -> Path:
return Path.home() / ".config" / "Moonlight Game Streaming Project"
def moonlight_state_file() -> str:
return os.path.join(moonlight_config_dir(), "Moonlight.conf")
def moonlight_state_file() -> Path:
return moonlight_config_dir() / "Moonlight.conf"
def load_state() -> ConfigParser | None:
try:
with open(moonlight_state_file()) as file:
with moonlight_state_file().open() as file:
config = ConfigParser()
config.read_file(file)
print(config.sections())
@@ -65,8 +64,8 @@ def init_state(certificate: str, key: str) -> None:
config.write(file)
def write_state(data: ConfigParser) -> bool:
with open(moonlight_state_file(), "w") as file:
def write_state(data: ConfigParser) -> None:
with moonlight_state_file().open("w") as file:
data.write(file)
@@ -84,22 +83,24 @@ def add_sunshine_host_to_parser(
new_host = no_hosts + 1
config.set("hosts", f"{new_host}\srvcert", convert_string_to_bytearray(certificate))
config.set(
"hosts", f"{new_host}\\srvcert", convert_string_to_bytearray(certificate)
)
config.set("hosts", "size", str(new_host))
config.set("hosts", f"{new_host}\\uuid", uuid)
config.set("hosts", f"{new_host}\hostname", hostname)
config.set("hosts", f"{new_host}\\hostname", hostname)
config.set("hosts", f"{new_host}\\nvidiasv", "false")
config.set("hosts", f"{new_host}\customname", "false")
config.set("hosts", f"{new_host}\manualaddress", manual_host)
config.set("hosts", f"{new_host}\manualport", "47989")
config.set("hosts", f"{new_host}\\customname", "false")
config.set("hosts", f"{new_host}\\manualaddress", manual_host)
config.set("hosts", f"{new_host}\\manualport", "47989")
config.set("hosts", f"{new_host}\\remoteport", "0")
config.set("hosts", f"{new_host}\\remoteaddress", "")
config.set("hosts", f"{new_host}\localaddress", "")
config.set("hosts", f"{new_host}\localport", "0")
config.set("hosts", f"{new_host}\ipv6port", "0")
config.set("hosts", f"{new_host}\ipv6address", "")
config.set("hosts", f"{new_host}\\localaddress", "")
config.set("hosts", f"{new_host}\\localport", "0")
config.set("hosts", f"{new_host}\\ipv6port", "0")
config.set("hosts", f"{new_host}\\ipv6address", "")
config.set(
"hosts", f"{new_host}\mac", convert_string_to_bytearray("\\xceop\\x8d\\xfc{")
"hosts", f"{new_host}\\mac", convert_string_to_bytearray("\\xceop\\x8d\\xfc{")
)
add_app(config, "Desktop", new_host, 1, 881448767)
add_app(config, "Low Res Desktop", new_host, 2, 303580669)
@@ -114,7 +115,7 @@ def add_sunshine_host_to_parser(
def add_app(
config: ConfigParser, name: str, host_id: int, app_id: int, app_no: int
) -> None:
identifier = f"{host_id}\\apps\{app_id}\\"
identifier = f"{host_id}\\apps\\{app_id}\\"
config.set("hosts", f"{identifier}appcollector", "false")
config.set("hosts", f"{identifier}directlaunch", "false")
config.set("hosts", f"{identifier}hdr", "false")

5
pkgs/moonlight-sunshine-accept/moonlight_sunshine_accept/moonlight/uri.py
View File

@@ -1,7 +1,7 @@
from urllib.parse import urlparse
def parse_moonlight_uri(uri: str) -> (str, str):
def parse_moonlight_uri(uri: str) -> tuple[str, int | None]:
print(uri)
if uri.startswith("moonlight:"):
# Fixes a bug where moonlight:// is not parsed correctly
@@ -13,5 +13,8 @@ def parse_moonlight_uri(uri: str) -> (str, str):
msg = f"Invalid moonlight URI: {uri}"
raise ValueError(msg)
hostname = parsed.hostname
if hostname is None:
msg = f"Invalid moonlight URI: {uri}"
raise ValueError
port = parsed.port
return (hostname, port)