Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion MAC/FindMyFlipperMac/Backend/findmy_gateway/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -245,7 +245,7 @@ def _second_factor_headers(module: ModuleType, dsid: str, idms_token: str) -> di
"X-Apple-Identity-Token": identity_token,
"X-Apple-App-Info": "com.apple.gs.xcode.auth",
"X-Xcode-Version": "11.2 (11B41)",
"X-Mme-Client-Info": "<MacBookPro18,3> <Mac OS X;13.4.1;22F8> <com.apple.AOSKit/282 (com.apple.dt.Xcode/3594.4.19)>",
"X-Mme-Client-Info": module.GSA_CLIENT_INFO,
}
headers.update(module.generate_anisette_headers())
return headers
Expand Down
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
from getpass import getpass
import plistlib as plist
import json
import logging
import time
import uuid
import pbkdf2
import requests
Expand Down Expand Up @@ -28,6 +30,11 @@
# urllib3.disable_warnings()

ANISETTE_URL = 'http://localhost:6969' # https://github.com/Dadoum/anisette-v3-server
GSA_CLIENT_INFO = '<MacBookPro18,3> <Mac OS X;13.4.1;22F8> <com.apple.AOSKit/282 (com.apple.akd/1.0)>'
GSA_RETRY_STATUS_CODES = {502, 503, 504}
GSA_MAX_ATTEMPTS = 3
GSA_RETRY_BACKOFF_SECONDS = 0.5
logger = logging.getLogger(__name__)


def icloud_login_mobileme(username='', password='', second_factor='sms'):
Expand Down Expand Up @@ -115,7 +122,8 @@ def gsa_authenticate(username, password, second_factor='sms'):
sms_second_factor(spd["adsid"], spd["GsIdmsToken"])
elif second_factor == 'trusted_device':
trusted_second_factor(spd["adsid"], spd["GsIdmsToken"])
return gsa_authenticate(username, password)
# Re-authenticate using the same 2FA method selected by the caller.
return gsa_authenticate(username, password, second_factor)
elif "au" in r["Status"]:
print(f"Unknown auth value {r['Status']['au']}")
return
Expand All @@ -134,18 +142,49 @@ def gsa_authenticated_request(parameters):
"Content-Type": "text/x-xml-plist",
"Accept": "*/*",
"User-Agent": "akd/1.0 CFNetwork/978.0.7 Darwin/18.7.0",
"X-MMe-Client-Info": '<MacBookPro18,3> <Mac OS X;13.4.1;22F8> <com.apple.AOSKit/282 (com.apple.dt.Xcode/3594.4.19)>'
# Apple started rejecting the legacy Xcode client identifier at the
# GrandSlam edge with a 190-byte HTML 503 response in September 2026.
"X-MMe-Client-Info": GSA_CLIENT_INFO,
}

resp = requests.post(
"https://gsa.apple.com/grandslam/GsService2",
headers=headers,
data=plist.dumps(body),
verify=False,
timeout=5,
)
request_body = plist.dumps(body)
for attempt in range(GSA_MAX_ATTEMPTS):
# requests.post() creates a fresh short-lived Session, so retries do
# not remain pinned to the same failing Apple edge connection.
resp = requests.post(
"https://gsa.apple.com/grandslam/GsService2",
headers=headers,
data=request_body,
verify=False,
timeout=5,
)
if resp.status_code not in GSA_RETRY_STATUS_CODES or attempt == GSA_MAX_ATTEMPTS - 1:
break

delay = GSA_RETRY_BACKOFF_SECONDS * (2 ** attempt)
logger.warning(
"Apple GSA returned HTTP %s; retrying in %.1f seconds (%s/%s)",
resp.status_code,
delay,
attempt + 1,
GSA_MAX_ATTEMPTS - 1,
)
time.sleep(delay)

try:
parsed_response = plist.loads(resp.content)
except (plist.InvalidFileException, TypeError, ValueError) as exc:
content_type = resp.headers.get("Content-Type", "<missing>")
content_preview = resp.content[:1000]
message = (
"Apple GSA returned an invalid plist response: "
f"HTTP {resp.status_code}; Content-Type: {content_type}; "
f"first {len(content_preview)} bytes: {content_preview!r}"
)
logger.error(message)
raise RuntimeError(message) from exc

return plist.loads(resp.content)["Response"]
return parsed_response["Response"]


def generate_cpd():
Expand All @@ -172,7 +211,7 @@ def generate_anisette_headers():
device = pyprovision.Device("./anisette/device.json")
if not device.initialized:
# Pretend to be a MacBook Pro
device.server_friendly_description = "<MacBookPro13,2> <macOS;13.1;22C65> <com.apple.AuthKit/1 (com.apple.dt.Xcode/3594.4.19)>"
device.server_friendly_description = "<MacBookPro13,2> <macOS;13.1;22C65> <com.apple.AuthKit/1 (com.apple.akd/1.0)>"
device.unique_device_identifier = str(uuid.uuid4()).upper()
device.adi_identifier = secrets.token_hex(8).lower()
device.local_user_uuid = secrets.token_hex(32).upper()
Expand All @@ -189,7 +228,27 @@ def generate_anisette_headers():
except ImportError:
print(f'pyprovision is not installed, querying {ANISETTE_URL} for an anisette server')
h = json.loads(requests.get(ANISETTE_URL, timeout=5).text)
a = {"X-Apple-I-MD": h["X-Apple-I-MD"], "X-Apple-I-MD-M": h["X-Apple-I-MD-M"]}
# Keep the complete device identity returned by anisette-v3. Its machine
# identifier is provisioned together with the device ID, local-user ID,
# serial number, and metadata below. Mixing it with locally generated
# UUIDs prevents Apple from trusting the device after successful 2FA.
return {
"X-Apple-I-MD": h["X-Apple-I-MD"],
"X-Apple-I-MD-M": h["X-Apple-I-MD-M"],
"X-Apple-I-MD-RINFO": h.get("X-Apple-I-MD-RINFO", "17106176"),
"X-Apple-I-MD-LU": h.get("X-Apple-I-MD-LU", ""),
"X-Mme-Device-Id": h.get("X-Mme-Device-Id", ""),
"X-Apple-I-SRL-NO": h.get("X-Apple-I-SRL-NO", "0"),
"X-Apple-I-Client-Time": h.get(
"X-Apple-I-Client-Time",
datetime.utcnow().replace(microsecond=0).isoformat() + "Z",
),
"X-Apple-I-TimeZone": h.get("X-Apple-I-TimeZone", "UTC"),
"loc": locale.getdefaultlocale()[0] or "en_US",
"X-Apple-Locale": h.get(
"X-Apple-Locale", locale.getdefaultlocale()[0] or "en_US"
),
}
a.update(generate_meta_headers(user_id=USER_ID, device_id=DEVICE_ID))
return a

Expand Down Expand Up @@ -246,7 +305,7 @@ def trusted_second_factor(dsid, idms_token):
"X-Apple-Identity-Token": identity_token,
"X-Apple-App-Info": "com.apple.gs.xcode.auth",
"X-Xcode-Version": "11.2 (11B41)",
"X-Mme-Client-Info": '<MacBookPro18,3> <Mac OS X;13.4.1;22F8> <com.apple.AOSKit/282 (com.apple.dt.Xcode/3594.4.19)>'
"X-Mme-Client-Info": GSA_CLIENT_INFO,
}

headers.update(generate_anisette_headers())
Expand Down Expand Up @@ -290,7 +349,7 @@ def sms_second_factor(dsid, idms_token):
"X-Apple-Identity-Token": identity_token,
"X-Apple-App-Info": "com.apple.gs.xcode.auth",
"X-Xcode-Version": "11.2 (11B41)",
"X-Mme-Client-Info": '<MacBookPro18,3> <Mac OS X;13.4.1;22F8> <com.apple.AOSKit/282 (com.apple.dt.Xcode/3594.4.19)>'
"X-Mme-Client-Info": GSA_CLIENT_INFO,
}

headers.update(generate_anisette_headers())
Expand Down
12 changes: 12 additions & 0 deletions MAC/FindMyFlipperMac/Backend/tests/test_auth_vendor.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,3 +90,15 @@ def fake_login(username, password, second_factor="sms"):
("user@example.com", "secret-password", "trusted_device"),
]
assert manager._pending_second_factor is None


def test_second_factor_headers_use_vendor_akd_client_identity():
module = types.SimpleNamespace(
GSA_CLIENT_INFO="client-info-with-com.apple.akd/1.0",
generate_anisette_headers=lambda: {},
)

headers = AuthManager._second_factor_headers(module, "123", "token")

assert headers["X-Mme-Client-Info"] == module.GSA_CLIENT_INFO
assert "com.apple.dt.Xcode" not in headers["X-Mme-Client-Info"]
164 changes: 164 additions & 0 deletions MAC/FindMyFlipperMac/Backend/tests/test_pypush_gsa_icloud.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,164 @@
import json

import pytest

from findmy_gateway.vendor.cores import pypush_gsa_icloud as gsa


def test_anisette_server_headers_keep_server_device_identity(monkeypatch):
server_headers = {
"X-Apple-I-MD": "otp",
"X-Apple-I-MD-M": "machine-id",
"X-Apple-I-MD-RINFO": "server-rinfo",
"X-Apple-I-MD-LU": "server-local-user",
"X-Mme-Device-Id": "server-device-id",
"X-Apple-I-SRL-NO": "server-serial",
"X-Apple-I-Client-Time": "2026-09-12T01:02:03Z",
"X-Apple-I-TimeZone": "Asia/Chita",
"X-Apple-Locale": "en_US",
}
response = type("Response", (), {"text": json.dumps(server_headers)})()

monkeypatch.setattr(gsa.requests, "get", lambda *args, **kwargs: response)
monkeypatch.setattr(gsa.locale, "getdefaultlocale", lambda: ("ru_RU", "UTF-8"))
monkeypatch.setattr(
gsa,
"generate_meta_headers",
lambda *args, **kwargs: pytest.fail("server headers must not be mixed with local UUIDs"),
)

assert gsa.generate_anisette_headers() == {
**server_headers,
"loc": "ru_RU",
}


def test_gsa_authenticate_keeps_second_factor_on_retry(monkeypatch):
class FakeUser:
def __init__(self, *args, **kwargs):
self.p = None

def start_authentication(self):
return None, b"A"

def process_challenge(self, salt, challenge):
return b"M1"

def verify_session(self, server_proof):
return None

def authenticated(self):
return True

complete_requests = 0

def fake_authenticated_request(parameters):
nonlocal complete_requests
if parameters["o"] == "init":
return {"sp": "s2k", "s": b"salt", "i": 1, "B": b"challenge", "c": "context"}

complete_requests += 1
status = {"au": "trustedDeviceSecondaryAuth"} if complete_requests == 1 else {}
return {"M2": b"proof", "spd": b"encrypted", "Status": status}

trusted_calls = []
sms_calls = []
monkeypatch.setattr(gsa.srp, "User", FakeUser)
monkeypatch.setattr(gsa, "gsa_authenticated_request", fake_authenticated_request)
monkeypatch.setattr(gsa, "encrypt_password", lambda *args, **kwargs: b"password")
monkeypatch.setattr(gsa, "decrypt_cbc", lambda *args, **kwargs: b"plist")
monkeypatch.setattr(
gsa.plist,
"loads",
lambda data: {"adsid": "123", "GsIdmsToken": "token"},
)
monkeypatch.setattr(gsa, "trusted_second_factor", lambda *args: trusted_calls.append(args))
monkeypatch.setattr(gsa, "sms_second_factor", lambda *args: sms_calls.append(args))

result = gsa.gsa_authenticate("user@example.com", "password", "trusted_device")

assert result == {"adsid": "123", "GsIdmsToken": "token"}
assert trusted_calls == [("123", "token")]
assert sms_calls == []
assert complete_requests == 2


def test_gsa_request_reports_non_plist_response_details(monkeypatch):
response = type(
"Response",
(),
{
"status_code": 503,
"headers": {"Content-Type": "text/html; charset=utf-8"},
"content": b"<html>" + (b"x" * 1100) + b"not-in-preview",
},
)()
monkeypatch.setattr(gsa.requests, "post", lambda *args, **kwargs: response)
monkeypatch.setattr(gsa, "generate_cpd", lambda: {})
monkeypatch.setattr(gsa.time, "sleep", lambda delay: None)

with pytest.raises(RuntimeError) as exc_info:
gsa.gsa_authenticated_request({"o": "init"})

message = str(exc_info.value)
assert "HTTP 503" in message
assert "Content-Type: text/html; charset=utf-8" in message
assert "first 1000 bytes" in message
assert "<html>" in message
assert "not-in-preview" not in message


def test_gsa_request_uses_unblocked_akd_client_identity(monkeypatch):
captured_headers = []
response = type(
"Response",
(),
{
"status_code": 200,
"headers": {"Content-Type": "text/x-xml-plist"},
"content": gsa.plist.dumps({"Response": {"ok": True}}),
},
)()

def fake_post(*args, **kwargs):
captured_headers.append(kwargs["headers"])
return response

monkeypatch.setattr(gsa.requests, "post", fake_post)
monkeypatch.setattr(gsa, "generate_cpd", lambda: {})

assert gsa.gsa_authenticated_request({"o": "init"}) == {"ok": True}
assert captured_headers[0]["X-MMe-Client-Info"] == gsa.GSA_CLIENT_INFO
assert "com.apple.akd/1.0" in gsa.GSA_CLIENT_INFO
assert "com.apple.dt.Xcode" not in gsa.GSA_CLIENT_INFO


def test_gsa_request_retries_transient_503_on_fresh_requests(monkeypatch):
responses = [
type(
"Response",
(),
{
"status_code": 503,
"headers": {"Content-Type": "text/html"},
"content": b"<html>temporary</html>",
},
)(),
type(
"Response",
(),
{
"status_code": 200,
"headers": {"Content-Type": "text/x-xml-plist"},
"content": gsa.plist.dumps({"Response": {"ok": True}}),
},
)(),
]
sleeps = []
monkeypatch.setattr(gsa.requests, "post", lambda *args, **kwargs: responses.pop(0))
monkeypatch.setattr(gsa, "generate_cpd", lambda: {})
monkeypatch.setattr(gsa.time, "sleep", sleeps.append)

assert gsa.gsa_authenticated_request({"o": "init"}) == {"ok": True}
assert responses == []
assert sleeps == [0.5]
Loading