yardım field 15 ekledim çalışmıyo

thorfinlol' Alıntı:

değil

Genişletmek için tıkla ...

sfdfdgffgfgdfgd' Alıntı:

import base64
import os
import hashlib
import json
import logging
from typing import Any, Dict
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat

logging.basicConfig(level=logging.INFO, format='%(message)s')
app = FastAPI()

# ============================================================================
# CONSTANTS
# ============================================================================

API_KEY = "fiverdbull"
RG_MAGIC = b"\x52\x47\x01\x00"

SERVER_RSA_PUBKEY_B64 = (
"MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAxABI8XC5dAnqbJX6ZLWjwiAPl18Pj/5q1E8I8raxRRJb7i7wmCuUEExKfwOBjbF4y4Wiugt8cloQniGqTzL7JfvjpaZWYuM7OTd9YjACJRmm6CDjNsAxAA6PzH7B9LJd49Cp4ViHME65uUorsnQ6riq0wbZSwNLaKWi9yoLlEX8Ru2CRgHJte35Fo1BcbB2S36SfwBu9tKMUbn1sAqjGMnzu8Slm9smtIoeugfvBEz4rpPpBH8n4/Nv89pZPf12O/64bHBfuK4v/g6Ig3T4MT73MmXfxP4Hv3pI9+9ydnkzc3uZ+4LZbONVCyjJHcndeOcTzw2VIcJP5gVAxgaW6WQIDAQAB"
)

VG_AUTH_REQUEST = 3

# ============================================================================
# PROTOBUF HELPERS
# ============================================================================

def _varint(n: int) -> bytes:
out = bytearray()
while True:
b = n & 0x7F
n >>= 7
if n:
b |= 0x80
out.append(b)
if not n:
break
return bytes(out)

def _field_bytes(field: int, data: bytes) -> bytes:
return _varint((field << 3) | 2) + _varint(len(data)) + data

def _field_str(field: int, s: str) -> bytes:
return _field_bytes(field, s.encode("utf-8"))

def _field_varint(field: int, v: int) -> bytes:
return _varint((field << 3) | 0) + _varint(v)

def _dev_sub(build: int = 36) -> bytes:
"""f6 ve f7: 1.18.4.36"""
b = bytearray()
b += _field_varint(1, 1)
b += _field_varint(2, 18)
b += _field_varint(3, 4)
b += _field_varint(4, build)
return bytes(b)

def _os_info(variant: int, arch: int, version: str) -> bytes:
"""
f2: {variant:6, arch:2, os_ver}
Field 1: variant (6)
Field 2: arch (2)
Field 3: os_ver (10.0.19045) <- FIELD 3 OLMALI!
"""
b = bytearray()
b += _field_varint(1, variant)
b += _field_varint(2, arch)
b += _field_str(3, version) # FIELD 3!
return bytes(b)

def _security_feature(name: str, state: int = 1) -> bytes:
"""f14: security feature"""
b = bytearray()
b += _field_str(1, name)
b += _field_varint(2, state)
return bytes(b)

def _build_kvcr_ht(hwid: bytes) -> str:
"""f15 ht value — base64(sha256(hwid)[:20])."""
return base64.b64encode(hashlib.sha256(hwid).digest()[:20]).decode("ascii")

def _write_map_entry(buf: bytearray, field: int, key: str, val: str) -> None:
"""map<string,string> entry — field 15 icin."""
entry = bytearray()
entry += _field_str(1, key) # f1 = key (string)
entry += _field_str(2, val) # f2 = value (string)
buf += _field_bytes(field, bytes(entry))

# ============================================================================
# AUTH ENVELOPE BUILDER
# ============================================================================

def normalize_client_rsa(spki_b64_or_der: str) -> bytes:
raw = (spki_b64_or_der or "").strip()
if raw.startswith("-----BEGIN"):
pub = serialization.load_pem_public_key(raw.encode())
der = pub.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
return base64.b64encode(der)
if raw.startswith("MIIB") or raw.startswith("MIG"):
return raw.encode()
return base64.b64encode(raw.encode())

def build_auth_plain(profile: Dict[str, Any]) -> bytes:
"""
Fields:
f1 : base64 (machine_id)
f2 : {variant:6, arch:2, os_ver}
f4 : JWT
f5 : client RSA SPKI
f6 : 1.18.4.36
f7 : 1.18.4.36
f8 : com.riotgames.valorant
f9 : 3
f13 : SID
f14 : ["HVCI","IOMMU","SB","VBS","TPM2"]
f15 : kvcr_ht map entry { "ht": base64(sha256(hwid)[:20]) }
"""
jwt = str(profile["jwt"])
external_sid = str(profile["external_sid"])
machine_id = str(profile["machine_id"])
client_rsa = normalize_client_rsa(profile["client_rsa"])
os_ver = str(profile.get("os_version", "10.0.19045"))
dev_build = int(profile.get("dev_build", 36))
boot_state = int(profile.get("boot_state", 3))
game_id = str(profile.get("game_id", "com.riotgames.valorant"))
hwid = profile.get("hwid")

sec = ["HVCI", "IOMMU", "SB", "VBS", "TPM2"]

buf = bytearray()
buf += _field_str(1, machine_id) # f1 - base64
buf += _field_bytes(2, _os_info(6, 2, os_ver)) # f2 - variant=6
buf += _field_str(4, jwt) # f4
buf += _field_bytes(5, client_rsa) # f5
buf += _field_bytes(6, _dev_sub(dev_build)) # f6
buf += _field_bytes(7, _dev_sub(dev_build)) # f7
buf += _field_str(8, game_id) # f8
buf += _field_varint(9, boot_state) # f9
buf += _field_str(13, external_sid) # f13
for feat in sec:
buf += _field_bytes(14, _security_feature(feat)) # f14

# f15 - kvcr_ht
kvcr_ht = str(profile.get("kvcr_ht") or "")
if not kvcr_ht and hwid:
kvcr_ht = _build_kvcr_ht(hwid)
if kvcr_ht:
_write_map_entry(buf, 15, "ht", kvcr_ht)

return bytes(buf)

def rg_encrypt(plaintext: bytes) -> bytes:
pub = serialization.load_der_public_key(base64.b64decode(SERVER_RSA_PUBKEY_B64))
aes_key = os.urandom(32)

oaep = padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA512()),
algorithm=hashes.SHA512(),
label=None,
)
enc_key = pub.encrypt(aes_key, oaep)
if len(enc_key) != 256:
raise ValueError(f"RSA block {len(enc_key)} != 256")

nonce = os.urandom(12)
ct = AESGCM(aes_key).encrypt(nonce, plaintext, None)

return RG_MAGIC + enc_key + nonce + ct

def build_envelope(rg_blob: bytes, msg_type: int = VG_AUTH_REQUEST) -> bytes:
buf = bytearray()
buf += _field_varint(1, msg_type)
buf += _field_bytes(2, rg_blob)
return bytes(buf)

def build_auth_http_body(profile: Dict[str, Any]) -> tuple[bytes, bytes]:
plain = build_auth_plain(profile)
rg = rg_encrypt(plain)
envelope = build_envelope(rg)
return plain, envelope

# ============================================================================
# EMU ENVELOPE
# ============================================================================

def derive_key(api_key: str) -> bytes:
return hashlib.sha256(api_key.encode("utf-8")).digest()

def decrypt_payload(api_key: str, encoded_b64: str) -> str:
raw = base64.b64decode(encoded_b64)
nonce = raw[:12]
ct_tag = raw[12:]
key = derive_key(api_key)
aesgcm = AESGCM(key)
plaintext = aesgcm.decrypt(nonce, ct_tag, None)
return plaintext.decode("utf-8")

def encrypt_payload(api_key: str, plaintext: str) -> str:
key = derive_key(api_key)
nonce = os.urandom(12)
aesgcm = AESGCM(key)
ct_tag = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None)
combined = nonce + ct_tag
return base64.b64encode(combined).decode("utf-8")

def make_response(api_key: str, success: bool, data: Dict) -> str:
resp = {"success": success}
resp.update(data)
return encrypt_payload(api_key, json.dumps(resp, separators=(",", ":")))

def open_envelope(request_json: Dict) -> Dict:
d = request_json.get("d", "")
if not d:
return request_json
decrypted = decrypt_payload(request_json.get("t", API_KEY), d)
return json.loads(decrypted)

# ============================================================================
# FASTAPI ENDPOINT
# ============================================================================

@app.post("/vgc/session/gateway")
async def gateway(request: Request):
try:
body = await request.json()
inner = open_envelope(body)

gametoken = inner.get("gametoken", "")
sid = inner.get("sid", "")
puuid = inner.get("puuid", "")

if not gametoken or not sid:
resp_json = make_response(API_KEY, False, {"error": "missing gametoken or sid"})
return JSONResponse(content={"d": resp_json}, status_code=400)

rsa_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
spki_b64 = base64.b64encode(
rsa_key.public_key().public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
).decode("ascii")

# f1 - base64
machine_id = base64.b64encode(os.urandom(32)).decode("ascii")

profile = {
"jwt": gametoken,
"external_sid": sid,
"machine_id": machine_id,
"client_rsa": spki_b64,
"os_version": "10.0.19045",
"dev_build": 36,
"boot_state": 3,
"game_id": "com.riotgames.valorant",
}

session_id = os.urandom(16).hex()
plain, envelope = build_auth_http_body(profile)
envelope_b64 = base64.b64encode(envelope).decode("ascii")

print(f"[GATEWAY] session_id: {session_id}")
print(f"[GATEWAY] f1: {machine_id[:30]}...")

resp_json = make_response(API_KEY, True, {
"data": envelope_b64,
"session_id": session_id,
"message": "auth envelope ready",
})
return JSONResponse(content={"d": resp_json})

except Exception as e:
print(f"[GATEWAY] exception: {e}")
import traceback
traceback.print_exc()
resp_json = make_response(API_KEY, False, {"error": str(e)})
return JSONResponse(content={"d": resp_json}, status_code=500)

@app.get("/health")
async def health():
return {"status": "ok"}

# ============================================================================
# MAIN
# ============================================================================

if __name__ == "__main__":
import uvicorn

print("=" * 60)
print(" Gateway Backend v3.0")
print("=" * 60)
print(" Port: 8000")
print(" Endpoint: /vgc/session/gateway")
print("=" * 60)
print("")
print(" Fields:")
print(" f1 : base64")
print(" f2 : {variant:6, arch:2, os_ver}")
print(" f4 : JWT")
print(" f5 : RSA SPKI")
print ("f6: 1.18.4.36")
print ("f7: 1.18.4.36")
print ("f8: com.riotgames.valorant")
print ("f9: 3")
print (" f13: SID")
print(" f14 : [HVCI, IOMMU, SB, VBS, TPM2]")
print(" f15 : kvcr_ht {ht: base64(sha256(hwid)[:20]) }")
print("=" * 60)
print("")

uvicorn.run(
app,
host="0.0.0.0",
port=8000,
log_level="info",
use_colors=False
) isn't it true

Genişletmek için tıkla ...

abdouthegoat06' Alıntı:

Incorrect value

Genişletmek için tıkla ...