Merge branch 'new_build' into dev

Author: danny

README.md

@@ -148,7 +148,7 @@ Included projects are (Credits goes to them for the hard work):
 
 Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.
 
-[![License: GPL v3](https://img.shields.io/badge/License-GPLv3-blue.svg)](https://www.gnu.org/licenses/gpl-3.0)
+[![License: AGPL v3](https://img.shields.io/badge/License-AGPL_v3-blue.svg)](https://www.gnu.org/licenses/agpl-3.0)
 
 ## Disclaimer
 

endpoint/config.py endpoint/mh_config.pyendpoint/config.py renamed to endpoint/mh_config.py

@@ -5,14 +5,15 @@
 import logging
 import os
 import ssl
+import sys
 import time
 from collections import OrderedDict
 from datetime import datetime,  timezone
 from http.server import BaseHTTPRequestHandler, HTTPServer
 
 import requests
 
-import config
+import mh_config
 from register import apple_cryptography, pypush_gsa_icloud
 
 logger = logging.getLogger()
@@ -29,9 +30,9 @@ def addCORSHeaders(self):
         self.send_header("Access-Control-Allow-Private-Network","true")
 
     def authenticate(self):
-        user = config.getEndpointUser()
-        passw = config.getEndpointPass()
-        if (user is None or user == "") and (passw is None or passw == ""):
+        endpoint_user = mh_config.getEndpointUser()
+        endpoint_pass = mh_config.getEndpointPass()
+        if (endpoint_user is None or endpoint_user == "") and (endpoint_pass is None or endpoint_pass == ""):
             return True
 
         auth_header = self.headers.get('authorization')
@@ -40,7 +41,7 @@ def authenticate(self):
             if auth_type.lower() == 'basic':
                 auth_decoded = base64.b64decode(auth_encoded).decode('utf-8')
                 username, password = auth_decoded.split(':', 1)
-                if username == user and password == passw:
+                if username == endpoint_user and password == endpoint_pass:
                     return True
 
         return False
@@ -122,7 +123,7 @@ def do_POST(self):
             responseBody = json.dumps(result)
             self.wfile.write(responseBody.encode())
         except requests.exceptions.ConnectTimeout:
-            logger.error("Timeout to " + config.getAnisetteServer() +
+            logger.error("Timeout to " + mh_config.getAnisetteServer() +
                          ", is your anisette running and accepting Connections?")
             self.send_response(504)
         except Exception as e:
@@ -136,46 +137,64 @@ def getCurrentTimes(self):
 
 
 def getAuth(regenerate=False, second_factor='sms'):
-    if os.path.exists(config.getConfigFile()) and not regenerate:
-        with open(config.getConfigFile(), "r") as f:
+    if os.path.exists(mh_config.getConfigFile()) and not regenerate:
+        with open(mh_config.getConfigFile(), "r") as f:
             j = json.load(f)
     else:
-        mobileme = pypush_gsa_icloud.icloud_login_mobileme(username=config.USER, password=config.PASS)
+        mobileme = pypush_gsa_icloud.icloud_login_mobileme(username=mh_config.USER, password=mh_config.PASS)
         logger.debug('Mobileme result: ' + mobileme)
         j = {'dsid': mobileme['dsid'], 'searchPartyToken': mobileme['delegates']
              ['com.apple.mobileme']['service-data']['tokens']['searchPartyToken']}
-        with open(config.getConfigFile(), "w") as f:
+        with open(mh_config.getConfigFile(), "w") as f:
             json.dump(j, f)
     return j['dsid'], j['searchPartyToken']
 
 
+
+def check_if_anisette_is_reachable(max_retries=3, retry_delay=10):
+    server_url = mh_config.getAnisetteServer()
+    logging.info(f'Checking if Anisette {server_url} is reachable')
+    for attempt in range(max_retries):
+        try:
+            response = requests.get(server_url, timeout=5)
+            response.raise_for_status()
+            return
+        except (requests.RequestException, requests.HTTPError) as e:
+            logger.error(f"Attempt {attempt + 1} failed: {str(e)}")
+            if attempt < max_retries - 1:
+                logger.error(f"Retrying in {retry_delay} seconds...")
+                time.sleep(retry_delay)
+    logger.error(f"Max retries reached. Program will exit. Make sure your Anisette is reachable and start again with 'docker start -ai macless-haystack'")
+    sys.exit()
+
 if __name__ == "__main__":
-    logging.debug(f'Searching for token at ' + config.getConfigFile())
-    if not os.path.exists(config.getConfigFile()):
+    check_if_anisette_is_reachable()
+    logging.info(f'Searching for token at ' + mh_config.getConfigFile())
+    if not os.path.exists(mh_config.getConfigFile()):
         logging.info(f'No auth-token found.')
         apple_cryptography.registerDevice()
 
     Handler = ServerHandler
 
-    httpd = HTTPServer((config.getBindingAddress(), config.getPort()), Handler)
+    httpd = HTTPServer((mh_config.getBindingAddress(), mh_config.getPort()), Handler)
     httpd.timeout = 30
-    address = config.getBindingAddress() + ":" + str(config.getPort())
-    if os.path.isfile(config.getCertFile()):
-        logger.info("Certificate file " + config.getCertFile() +
+    address = mh_config.getBindingAddress() + ":" + str(mh_config.getPort())
+    if os.path.isfile(mh_config.getCertFile()):
+        logger.info("Certificate file " + mh_config.getCertFile() +
                     " exists, so using SSL")
         ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
-        ssl_context.load_cert_chain(certfile=config.getCertFile(
-        ), keyfile=config.getKeyFile() if os.path.isfile(config.getKeyFile()) else None)
+        ssl_context.load_cert_chain(certfile=mh_config.getCertFile(
+        ), keyfile=mh_config.getKeyFile() if os.path.isfile(mh_config.getKeyFile()) else None)
 
         httpd.socket = ssl_context.wrap_socket(httpd.socket, server_side=True)
 
         logger.info("serving at " + address + " over HTTPS")
     else:
-        logger.info("Certificate file " + config.getCertFile() +
+        logger.info("Certificate file " + mh_config.getCertFile() +
                     " not found, so not using SSL")
         logger.info("serving at " + address + " over HTTP")
-    user = config.getEndpointUser()
-    passw = config.getEndpointPass()
+    user = mh_config.getEndpointUser()
+    passw = mh_config.getEndpointPass()
     if (user is None or user == "") and (passw is None or passw == ""):
         logger.warning("Endpoint is not protected by authentication")
     else:

endpoint/mh_endpoint.py

@@ -1,24 +1,18 @@
 #!/usr/bin/env python
 
-import os
-import glob
-import datetime
-import argparse
-import base64
-import json
 import hashlib
-import codecs
+import json
+import logging
+import os
 import struct
-import requests
-from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
-from cryptography.hazmat.primitives.asymmetric import ec
-from cryptography.hazmat.backends import default_backend
-import sqlite3
 import sys
-from .pypush_gsa_icloud import icloud_login_mobileme, generate_anisette_headers
 
-import config
-import logging
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives.ciphers import Cipher
+
+from endpoint import mh_config
+from .pypush_gsa_icloud import icloud_login_mobileme
+
 logger = logging.getLogger()
 
 
@@ -42,21 +36,21 @@ def decode_tag(data):
 
 
 def getAuth(regenerate=False):
-    if os.path.exists(config.getConfigFile()) and not regenerate:
-        with open(config.getConfigFile(), "r") as f:
+    if os.path.exists(mh_config.getConfigFile()) and not regenerate:
+        with open(mh_config.getConfigFile(), "r") as f:
             j = json.load(f)
     else:
         logger.info('Trying to login')
         mobileme = icloud_login_mobileme(
-            username=config.getUser(), password=config.getPass())
+            username=mh_config.getUser(), password=mh_config.getPass())
 
         logger.debug('Answer from icloud login')
         logger.debug(mobileme)
         status = mobileme['delegates']['com.apple.mobileme']['status']
         if status == 0:
             j = {'dsid': mobileme['dsid'], 'searchPartyToken': mobileme['delegates']
                  ['com.apple.mobileme']['service-data']['tokens']['searchPartyToken']}
-            with open(config.getConfigFile(), "w") as f:
+            with open(mh_config.getConfigFile(), "w") as f:
                 json.dump(j, f)
         else:
             msg = mobileme['delegates']['com.apple.mobileme']['status-message']

endpoint/register/apple_cryptography.py

@@ -1,3 +1,5 @@
+import time
+
 import urllib3
 from getpass import getpass
 import plistlib as plist
@@ -16,7 +18,8 @@
 from cryptography.hazmat.primitives import padding
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 from Crypto.Hash import SHA256
-import config
+
+from endpoint import mh_config
 
 # Created here so that it is consistent
 USER_ID = uuid.uuid4()
@@ -37,6 +40,7 @@ def icloud_login_mobileme(username='', password=''):
         username = input('Apple ID: ')
     if not password:
         password = getpass('Password: ')
+
     g = gsa_authenticate(username, password)
     pet = g["t"]["com.apple.gs.idms.pet"]["token"]
     adsid = g["adsid"]
@@ -71,10 +75,10 @@ def icloud_login_mobileme(username='', password=''):
 def gsa_authenticate(username, password):
     # Password is None as we'll provide it later
     usr = srp.User(username, bytes(), hash_alg=srp.SHA256, ng_type=srp.NG_2048)
-    _, A = usr.start_authentication()
+    _, a = usr.start_authentication()
     logger.info("Authentication request initialization")
     r = gsa_authenticated_request(
-        {"A2k": A, "ps": ["s2k", "s2k_fo"], "u": username, "o": "init"})
+        {"A2k": a, "ps": ["s2k", "s2k_fo"], "u": username, "o": "init"})
 
     if r["sp"] not in ["s2k", "s2k_fo"]:
         logger.warning(f"This implementation only supports s2k and sk2_fo. Server returned {r['sp']}")
@@ -169,7 +173,7 @@ def generate_cpd():
 
 
 def generate_anisette_headers():
-    h = json.loads(requests.get(config.getAnisetteServer(), timeout=5).text)
+    h = json.loads(requests.get(mh_config.getAnisetteServer(), timeout=5).text)
     a = {"X-Apple-I-MD": h["X-Apple-I-MD"],
          "X-Apple-I-MD-M": h["X-Apple-I-MD-M"]}
     a.update(generate_meta_headers(user_id=USER_ID, device_id=DEVICE_ID))
@@ -219,15 +223,13 @@ def decrypt_cbc(usr, data):
     return padder.update(data) + padder.finalize()
 
 
+WAITING_TIME = 60
+
+
 def sms_second_factor(dsid, idms_token):
     identity_token = base64.b64encode(
         (dsid + ":" + idms_token).encode()).decode()
 
-    # TODO: Actually do this request to get user prompt data
-    # a = requests.get("https://gsa.apple.com/auth", verify=False)
-    # This request isn't strictly necessary though,
-    # and most accounts should have their id 1 SMS, if not contribute ;)
-
     headers = {
         "User-Agent": "Xcode",
         "Accept-Language": "en-us",
@@ -238,32 +240,46 @@ def sms_second_factor(dsid, idms_token):
     }
 
     headers.update(generate_anisette_headers())
-    
+
     # Extract the "boot_args" from the auth page to get the id of the trusted phone number
     pattern = r'<script.*class="boot_args">\s*(.*?)\s*</script>'
     auth = requests.get("https://gsa.apple.com/auth", headers=headers, verify=False)
     sms_id = 1
     match = re.search(pattern, auth.text, re.DOTALL)
     if match:
         boot_args = json.loads(match.group(1).strip())
-        try: 
+        try:
             sms_id = boot_args["direct"]["phoneNumberVerification"]["trustedPhoneNumber"]["id"]
-        except KeyError as e:    
-            logger.debug(match.group(1).strip())        
-            logger.error("Key for sms id not found. Using the first phone number")        
+        except KeyError as e:
+            logger.debug(match.group(1).strip())
+            logger.error("Key for sms id not found. Using the first phone number")
     else:
-      logger.debug(auth.text)
-      logger.error("Script for sms id not found. Using the first phone number")        
-        
-    logger.info(f"Using phone with id {sms_id} for SMS2FA")
-    body = {"phoneNumber": {"id": sms_id }, "mode": "sms"}
-
-    # This will send the 2FA code to the user's phone over SMS
-    # We don't care about the response, it's just some HTML with a form for entering the code
-    # Easier to just use a text prompt
+        logger.debug(auth.text)
+        logger.error("Script for sms id not found. Using the first phone number")
 
+    logger.info(f"Using phone with id {sms_id} for SMS2FA")
+    body = {"phoneNumber": {"id": sms_id}, "mode": "sms"}
+    for handler in logger.handlers:
+        handler.flush()
     # Prompt for the 2FA code. It's just a string like '123456', no dashes or spaces
-    code = input("Enter SMS 2FA code: ")
+    start_time = time.perf_counter()
+    code = input(
+        f"Enter SMS 2FA code (If you do not receive a code, wait {WAITING_TIME}s and press Enter. An attempt will be made to request the SMS in another way.): ")
+    end_time = time.perf_counter()
+
+    if code == "":
+        elapsed_time = int(end_time - start_time)
+        if elapsed_time < WAITING_TIME:
+            waiting_time = WAITING_TIME - elapsed_time
+            logger.info(
+                f"You only waited {elapsed_time} seconds. The next request will be started in {waiting_time} seconds")
+            time.sleep(waiting_time)
+            code = input(f"Enter SMS 2FA code if you have received it in the meantime, otherwise press Enter: ")
+
+            if code == "":
+                code = request_code(headers)
+        else:
+            code = request_code(headers)
 
     body['securityCode'] = {'code': code}
 
@@ -287,3 +303,19 @@ def sms_second_factor(dsid, idms_token):
     else:
         raise Exception(
             "2FA unsuccessful. Maybe wrong code or wrong number. Check your account details.")
+
+
+def request_code(headers):
+    # This will send the 2FA code to the user's phone over SMS
+    # We don't care about the response, it's just some HTML with a form for entering the code
+    # Easier to just use a text prompt
+    body = {"phoneNumber": {"id": 1}, "mode": "sms"}
+    requests.put(
+        "https://gsa.apple.com/auth/verify/phone/",
+        json=body,
+        headers=headers,
+        verify=False,
+        timeout=5
+    )
+    code = input(f"Enter SMS 2FA code:")
+    return  code

endpoint/register/pypush_gsa_icloud.py

@@ -1,3 +1,8 @@
+plugins {
+    id "com.android.application"
+    id "kotlin-android"
+    id "dev.flutter.flutter-gradle-plugin"
+}
 def localProperties = new Properties()
 def localPropertiesFile = rootProject.file('local.properties')
 if (localPropertiesFile.exists()) {
@@ -6,10 +11,6 @@ if (localPropertiesFile.exists()) {
     }
 }
 
-def flutterRoot = localProperties.getProperty('flutter.sdk')
-if (flutterRoot == null) {
-    throw new GradleException("Flutter SDK not found. Define location with flutter.sdk in the local.properties file.")
-}
 
 def flutterVersionCode = localProperties.getProperty('flutter.versionCode')
 if (flutterVersionCode == null) {
@@ -21,9 +22,6 @@ if (flutterVersionName == null) {
     flutterVersionName = '1.0'
 }
 
-apply plugin: 'com.android.application'
-apply plugin: 'kotlin-android'
-apply from: "$flutterRoot/packages/flutter_tools/gradle/flutter.gradle"
 
 android {
     defaultConfig {
@@ -76,5 +74,5 @@ flutter {
 }
 
 dependencies {
-    implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk7:$kotlin_version"
+    implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk7:1.8.21"
 }

macless_haystack/android/app/build.gradle

@@ -1,16 +1,3 @@
-buildscript {
-    ext.kotlin_version = '1.8.21'
-    repositories {
-        google()
-        mavenCentral()
-    }
-
-    dependencies {
-        classpath 'com.android.tools.build:gradle:8.6.1'
-        classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlin_version"
-    }
-}
-
 allprojects {
     repositories {
         google()