Первая публикация
This commit is contained in:
@@ -0,0 +1,6 @@
|
||||
"""The iCloudPy library."""
|
||||
import logging
|
||||
|
||||
from icloudpy.base import ICloudPyService # pylint: disable=unused-import
|
||||
|
||||
logging.getLogger(__name__).addHandler(logging.NullHandler())
|
||||
@@ -0,0 +1,598 @@
|
||||
"""Library base file."""
|
||||
import getpass
|
||||
import http.cookiejar as cookielib
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
from os import mkdir, path
|
||||
from re import match
|
||||
from tempfile import gettempdir
|
||||
from uuid import uuid1
|
||||
|
||||
from requests import Session
|
||||
from six import PY2
|
||||
|
||||
from icloudpy.exceptions import (
|
||||
ICloudPy2SARequiredException,
|
||||
ICloudPyAPIResponseException,
|
||||
ICloudPyFailedLoginException,
|
||||
ICloudPyServiceNotActivatedException,
|
||||
)
|
||||
from icloudpy.services import (
|
||||
DriveService,
|
||||
WorkService
|
||||
)
|
||||
from icloudpy.utils import get_password_from_keyring
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
HEADER_DATA = {
|
||||
"X-Apple-ID-Account-Country": "account_country",
|
||||
"X-Apple-ID-Session-Id": "session_id",
|
||||
"X-Apple-Session-Token": "session_token",
|
||||
"X-Apple-TwoSV-Trust-Token": "trust_token",
|
||||
"scnt": "scnt",
|
||||
}
|
||||
|
||||
|
||||
class ICloudPyPasswordFilter(logging.Filter):
|
||||
"""Password log hider."""
|
||||
|
||||
def __init__(self, password):
|
||||
super().__init__(password)
|
||||
|
||||
def filter(self, record):
|
||||
message = record.getMessage()
|
||||
if self.name in message:
|
||||
record.msg = message.replace(self.name, "*" * 8)
|
||||
record.args = []
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class ICloudPySession(Session):
|
||||
"""iCloud session."""
|
||||
|
||||
def __init__(self, service):
|
||||
self.service = service
|
||||
Session.__init__(self)
|
||||
|
||||
def request(self, method, url, **kwargs): # pylint: disable=arguments-differ
|
||||
|
||||
# Charge logging to the right service endpoint
|
||||
callee = inspect.stack()[2]
|
||||
module = inspect.getmodule(callee[0])
|
||||
request_logger = logging.getLogger(module.__name__).getChild("http")
|
||||
if self.service.password_filter not in request_logger.filters:
|
||||
request_logger.addFilter(self.service.password_filter)
|
||||
|
||||
request_logger.debug(f"{method} {url} {kwargs.get('data', '')}")
|
||||
|
||||
has_retried = kwargs.get("retried")
|
||||
kwargs.pop("retried", None)
|
||||
response = super().request(method, url, **kwargs)
|
||||
|
||||
content_type = response.headers.get("Content-Type", "").split(";")[0]
|
||||
json_mimetypes = ["application/json", "text/json"]
|
||||
|
||||
for header, value in HEADER_DATA.items():
|
||||
if response.headers.get(header):
|
||||
session_arg = value
|
||||
self.service.session_data.update(
|
||||
{session_arg: response.headers.get(header)}
|
||||
)
|
||||
|
||||
# Save session_data to file
|
||||
with open(self.service.session_path, "w", encoding="utf-8") as outfile:
|
||||
json.dump(self.service.session_data, outfile)
|
||||
LOGGER.debug("Saved session data to file")
|
||||
|
||||
# Save cookies to file
|
||||
if kwargs.get('save_cookie', True):
|
||||
self.cookies.save(ignore_discard=True, ignore_expires=True)
|
||||
LOGGER.debug("Cookies saved to %s", self.service.cookiejar_path)
|
||||
# print(response.text)
|
||||
if not response.ok and (
|
||||
content_type not in json_mimetypes
|
||||
or response.status_code in [421, 450, 500]
|
||||
):
|
||||
try:
|
||||
# pylint: disable=W0212
|
||||
fmip_url = self.service._get_webservice_url("findme")
|
||||
if (
|
||||
has_retried is None
|
||||
and response.status_code == 450
|
||||
and fmip_url in url
|
||||
):
|
||||
# Handle re-authentication for Find My iPhone
|
||||
LOGGER.debug("Re-authenticating Find My iPhone service")
|
||||
try:
|
||||
self.service.authenticate(True, "find")
|
||||
except ICloudPyAPIResponseException:
|
||||
LOGGER.debug("Re-authentication failed")
|
||||
kwargs["retried"] = True
|
||||
return self.request(method, url, **kwargs)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if has_retried is None and response.status_code in [421, 450, 500]:
|
||||
api_error = ICloudPyAPIResponseException(
|
||||
response.reason, response.status_code, retry=True
|
||||
)
|
||||
request_logger.debug(api_error)
|
||||
kwargs["retried"] = True
|
||||
return self.request(method, url, **kwargs)
|
||||
|
||||
self._raise_error(response.status_code, response.reason)
|
||||
|
||||
if content_type not in json_mimetypes:
|
||||
return response
|
||||
|
||||
try:
|
||||
data = response.json()
|
||||
except: # pylint: disable=bare-except
|
||||
request_logger.warning("Failed to parse response with JSON mimetype")
|
||||
return response
|
||||
|
||||
request_logger.debug(data)
|
||||
|
||||
if isinstance(data, dict):
|
||||
reason = data.get("errorMessage")
|
||||
reason = reason or data.get("reason")
|
||||
reason = reason or data.get("errorReason")
|
||||
if not reason and isinstance(data.get("error"), str):
|
||||
reason = data.get("error")
|
||||
if not reason and data.get("error"):
|
||||
reason = "Unknown reason"
|
||||
|
||||
code = data.get("errorCode")
|
||||
if not code and data.get("serverErrorCode"):
|
||||
code = data.get("serverErrorCode")
|
||||
|
||||
if reason:
|
||||
self._raise_error(code, reason)
|
||||
|
||||
return response
|
||||
|
||||
def _raise_error(self, code, reason):
|
||||
if (
|
||||
self.service.requires_2sa
|
||||
and reason == "Missing X-APPLE-WEBAUTH-TOKEN cookie"
|
||||
):
|
||||
raise ICloudPy2SARequiredException(self.service.user["apple_id"])
|
||||
if code in ("ZONE_NOT_FOUND", "AUTHENTICATION_FAILED"):
|
||||
reason = (
|
||||
reason + ". Please log into https://icloud.com/ to manually "
|
||||
"finish setting up your iCloud service"
|
||||
)
|
||||
api_error = ICloudPyServiceNotActivatedException(reason, code)
|
||||
LOGGER.error(api_error)
|
||||
|
||||
raise api_error
|
||||
if code == "ACCESS_DENIED":
|
||||
reason = (
|
||||
reason + ". Please wait a few minutes then try again."
|
||||
"The remote servers might be trying to throttle requests."
|
||||
)
|
||||
if code in [421, 450, 500]:
|
||||
reason = "Authentication required for Account."
|
||||
|
||||
|
||||
api_error = ICloudPyAPIResponseException(reason, code)
|
||||
print(api_error)
|
||||
LOGGER.error(api_error)
|
||||
raise api_error
|
||||
|
||||
# Public method to resolve linting error
|
||||
def raise_error(self, code, reason):
|
||||
return self._raise_error(code=code, reason=reason)
|
||||
|
||||
|
||||
class ICloudPyService:
|
||||
"""
|
||||
A base authentication class for the iCloud service. Handles the
|
||||
authentication required to access iCloud services.
|
||||
|
||||
Usage:
|
||||
from src import ICloudPyService
|
||||
icloudpy = ICloudPyService('username@apple.com', 'password')
|
||||
icloudpy.iphone.location()
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
apple_id,
|
||||
password=None,
|
||||
cookie_directory=None,
|
||||
verify=True,
|
||||
client_id=None,
|
||||
with_family=True,
|
||||
auth_endpoint="https://idmsa.apple.com/appleauth/auth",
|
||||
# For China, use "https://www.icloud.com.cn"
|
||||
home_endpoint="https://www.icloud.com",
|
||||
# For China, use "https://setup.icloud.com.cn/setup/ws/1"
|
||||
setup_endpoint="https://setup.icloud.com/setup/ws/1",
|
||||
):
|
||||
if password is None:
|
||||
password = get_password_from_keyring(apple_id)
|
||||
|
||||
self.user = {"accountName": apple_id, "password": password}
|
||||
self.data = {}
|
||||
self.params = {}
|
||||
self.client_id = client_id or (f"auth-{str(uuid1()).lower()}")
|
||||
self.with_family = with_family
|
||||
self.auth_endpoint = auth_endpoint
|
||||
self.home_endpoint = home_endpoint
|
||||
self.setup_endpoint = setup_endpoint
|
||||
|
||||
self.password_filter = ICloudPyPasswordFilter(password)
|
||||
LOGGER.addFilter(self.password_filter)
|
||||
|
||||
if cookie_directory:
|
||||
self._cookie_directory = path.expanduser(path.normpath(cookie_directory))
|
||||
if not path.exists(self._cookie_directory):
|
||||
mkdir(self._cookie_directory, 0o700)
|
||||
else:
|
||||
topdir = path.join(gettempdir(), "icloudpy")
|
||||
self._cookie_directory = path.join(topdir, getpass.getuser())
|
||||
if not path.exists(topdir):
|
||||
mkdir(topdir, 0o777)
|
||||
if not path.exists(self._cookie_directory):
|
||||
mkdir(self._cookie_directory, 0o700)
|
||||
|
||||
LOGGER.debug("Using session file %s", self.session_path)
|
||||
|
||||
self.session_data = {}
|
||||
try:
|
||||
with open(self.session_path, encoding="utf-8") as session_f:
|
||||
self.session_data = json.load(session_f)
|
||||
except: # pylint: disable=bare-except
|
||||
LOGGER.info("Session file does not exist")
|
||||
if self.session_data.get("client_id"):
|
||||
self.client_id = self.session_data.get("client_id")
|
||||
self.params["clientId"] = self.client_id
|
||||
else:
|
||||
self.session_data.update({"client_id": self.client_id})
|
||||
self.params["clientId"] = self.client_id
|
||||
|
||||
self.session = ICloudPySession(self)
|
||||
self.session.verify = verify
|
||||
self.session.headers.update(
|
||||
{"Origin": self.home_endpoint, "Referer": f"{self.home_endpoint}/"}
|
||||
)
|
||||
|
||||
cookiejar_path = self.cookiejar_path
|
||||
self.session.cookies = cookielib.LWPCookieJar(filename=cookiejar_path)
|
||||
if path.exists(cookiejar_path):
|
||||
try:
|
||||
self.session.cookies.load(ignore_discard=True, ignore_expires=True)
|
||||
LOGGER.debug("Read cookies from %s", cookiejar_path)
|
||||
except: # pylint: disable=bare-except
|
||||
# Most likely a pickled cookiejar from earlier versions.
|
||||
# The cookiejar will get replaced with a valid one after
|
||||
# successful authentication.
|
||||
LOGGER.warning("Failed to read cookiejar %s", cookiejar_path)
|
||||
|
||||
self.authenticate()
|
||||
|
||||
self._drive = None
|
||||
self._work = None
|
||||
self._files = None
|
||||
self._photos = None
|
||||
|
||||
def authenticate(self, force_refresh=False, service=None):
|
||||
"""
|
||||
Handles authentication, and persists cookies so that
|
||||
subsequent logins will not cause additional e-mails from Apple.
|
||||
"""
|
||||
|
||||
login_successful = False
|
||||
if self.session_data.get("session_token") and not force_refresh:
|
||||
LOGGER.debug("Checking session token validity")
|
||||
try:
|
||||
self.data = self._validate_token()
|
||||
login_successful = True
|
||||
except ICloudPyAPIResponseException:
|
||||
LOGGER.debug("Invalid authentication token, will log in from scratch.")
|
||||
|
||||
if not login_successful and service is not None:
|
||||
app = self.data["apps"][service]
|
||||
if (
|
||||
"canLaunchWithOneFactor" in app
|
||||
and app["canLaunchWithOneFactor"] is True
|
||||
):
|
||||
LOGGER.debug(
|
||||
"Authenticating as %s for %s", self.user["accountName"], service
|
||||
)
|
||||
|
||||
try:
|
||||
self._authenticate_with_credentials_service(service)
|
||||
login_successful = True
|
||||
except Exception as error:
|
||||
LOGGER.debug(
|
||||
"Could not log into service. Attempting brand new login. %s",
|
||||
str(error),
|
||||
)
|
||||
|
||||
if not login_successful:
|
||||
LOGGER.debug("Authenticating as %s", self.user["accountName"])
|
||||
|
||||
data = dict(self.user)
|
||||
|
||||
data["rememberMe"] = True
|
||||
data["trustTokens"] = []
|
||||
if self.session_data.get("trust_token"):
|
||||
data["trustTokens"] = [self.session_data.get("trust_token")]
|
||||
|
||||
headers = self._get_auth_headers()
|
||||
|
||||
if self.session_data.get("scnt"):
|
||||
headers["scnt"] = self.session_data.get("scnt")
|
||||
|
||||
if self.session_data.get("session_id"):
|
||||
headers["X-Apple-ID-Session-Id"] = self.session_data.get("session_id")
|
||||
|
||||
try:
|
||||
self.session.post(
|
||||
f"{self.auth_endpoint}/signin",
|
||||
params={"isRememberMeEnabled": "true"},
|
||||
data=json.dumps(data),
|
||||
headers=headers,
|
||||
)
|
||||
except ICloudPyAPIResponseException as error:
|
||||
msg = "Invalid email/password combination."
|
||||
raise ICloudPyFailedLoginException(msg, error) from error
|
||||
|
||||
self._authenticate_with_token()
|
||||
|
||||
self._webservices = self.data["webservices"]
|
||||
|
||||
LOGGER.debug("Authentication completed successfully")
|
||||
|
||||
def _authenticate_with_token(self):
|
||||
"""Authenticate using session token."""
|
||||
data = {
|
||||
"accountCountryCode": self.session_data.get("account_country"),
|
||||
"dsWebAuthToken": self.session_data.get("session_token"),
|
||||
"extended_login": True,
|
||||
"trustToken": self.session_data.get("trust_token", ""),
|
||||
}
|
||||
|
||||
try:
|
||||
req = self.session.post(
|
||||
f"{self.setup_endpoint}/accountLogin", data=json.dumps(data)
|
||||
)
|
||||
self.data = req.json()
|
||||
except ICloudPyAPIResponseException as error:
|
||||
msg = "Invalid authentication token."
|
||||
raise ICloudPyFailedLoginException(msg, error) from error
|
||||
|
||||
def _authenticate_with_credentials_service(self, service):
|
||||
"""Authenticate to a specific service using credentials."""
|
||||
data = {
|
||||
"appName": service,
|
||||
"apple_id": self.user["accountName"],
|
||||
"password": self.user["password"],
|
||||
}
|
||||
|
||||
try:
|
||||
self.session.post(
|
||||
f"{self.setup_endpoint}/accountLogin", data=json.dumps(data)
|
||||
)
|
||||
|
||||
self.data = self._validate_token()
|
||||
except ICloudPyAPIResponseException as error:
|
||||
msg = "Invalid email/password combination."
|
||||
raise ICloudPyFailedLoginException(msg, error) from error
|
||||
|
||||
def _validate_token(self):
|
||||
"""Checks if the current access token is still valid."""
|
||||
LOGGER.debug("Checking session token validity")
|
||||
try:
|
||||
req = self.session.post(f"{self.setup_endpoint}/validate", data="null")
|
||||
LOGGER.debug("Session token is still valid")
|
||||
return req.json()
|
||||
except ICloudPyAPIResponseException as err:
|
||||
LOGGER.debug("Invalid authentication token")
|
||||
raise err
|
||||
|
||||
def _get_auth_headers(self, overrides=None):
|
||||
headers = {
|
||||
"Accept": "*/*",
|
||||
"Content-Type": "application/json",
|
||||
"X-Apple-OAuth-Client-Id": "d39ba9916b7251055b22c7f910e2ea796ee65e98b2ddecea8f5dde8d9d1a815d",
|
||||
"X-Apple-OAuth-Client-Type": "firstPartyAuth",
|
||||
"X-Apple-OAuth-Redirect-URI": "https://www.icloud.com",
|
||||
"X-Apple-OAuth-Require-Grant-Code": "true",
|
||||
"X-Apple-OAuth-Response-Mode": "web_message",
|
||||
"X-Apple-OAuth-Response-Type": "code",
|
||||
"X-Apple-OAuth-State": self.client_id,
|
||||
"X-Apple-Widget-Key": "d39ba9916b7251055b22c7f910e2ea796ee65e98b2ddecea8f5dde8d9d1a815d",
|
||||
}
|
||||
if overrides:
|
||||
headers.update(overrides)
|
||||
return headers
|
||||
|
||||
@property
|
||||
def cookiejar_path(self):
|
||||
"""Get path for cookiejar file."""
|
||||
return path.join(
|
||||
self._cookie_directory,
|
||||
"".join([c for c in self.user.get("accountName") if match(r"\w", c)]),
|
||||
)
|
||||
|
||||
@property
|
||||
def session_path(self):
|
||||
"""Get path for session data file."""
|
||||
return path.join(
|
||||
self._cookie_directory,
|
||||
"".join([c for c in self.user.get("accountName") if match(r"\w", c)])
|
||||
+ ".session",
|
||||
)
|
||||
|
||||
@property
|
||||
def requires_2sa(self):
|
||||
"""Returns True if two-step authentication is required."""
|
||||
return self.data.get("dsInfo", {}).get("hsaVersion", 0) >= 1 and (
|
||||
self.data.get("hsaChallengeRequired", False) or not self.is_trusted_session
|
||||
)
|
||||
|
||||
@property
|
||||
def requires_2fa(self):
|
||||
"""Returns True if two-factor authentication is required."""
|
||||
return self.data["dsInfo"].get("hsaVersion", 0) == 2 and (
|
||||
self.data.get("hsaChallengeRequired", False) or not self.is_trusted_session
|
||||
)
|
||||
|
||||
@property
|
||||
def is_trusted_session(self):
|
||||
"""Returns True if the session is trusted."""
|
||||
return self.data.get("hsaTrustedBrowser", False)
|
||||
|
||||
@property
|
||||
def trusted_devices(self):
|
||||
"""Returns devices trusted for two-step authentication."""
|
||||
request = self.session.get(
|
||||
f"{self.setup_endpoint}/listDevices", params=self.params
|
||||
)
|
||||
return request.json().get("devices")
|
||||
|
||||
def send_verification_code(self, device):
|
||||
"""Requests that a verification code is sent to the given device."""
|
||||
data = json.dumps(device)
|
||||
request = self.session.post(
|
||||
f"{self.setup_endpoint}/sendVerificationCode",
|
||||
params=self.params,
|
||||
data=data,
|
||||
)
|
||||
return request.json().get("success", False)
|
||||
|
||||
def validate_verification_code(self, device, code):
|
||||
"""Verifies a verification code received on a trusted device."""
|
||||
device.update({"verificationCode": code, "trustBrowser": True})
|
||||
data = json.dumps(device)
|
||||
|
||||
try:
|
||||
self.session.post(
|
||||
f"{self.setup_endpoint}/validateVerificationCode",
|
||||
params=self.params,
|
||||
data=data,
|
||||
)
|
||||
except ICloudPyAPIResponseException as error:
|
||||
if error.code == -21669:
|
||||
# Wrong verification code
|
||||
return False
|
||||
raise
|
||||
|
||||
self.trust_session()
|
||||
|
||||
return not self.requires_2sa
|
||||
|
||||
def validate_2fa_code(self, code):
|
||||
"""Verifies a verification code received via Apple's 2FA system (HSA2)."""
|
||||
data = {"securityCode": {"code": code}}
|
||||
|
||||
headers = self._get_auth_headers({"Accept": "application/json"})
|
||||
|
||||
if self.session_data.get("scnt"):
|
||||
headers["scnt"] = self.session_data.get("scnt")
|
||||
|
||||
if self.session_data.get("session_id"):
|
||||
headers["X-Apple-ID-Session-Id"] = self.session_data.get("session_id")
|
||||
|
||||
try:
|
||||
self.session.post(
|
||||
f"{self.auth_endpoint}/verify/trusteddevice/securitycode",
|
||||
data=json.dumps(data),
|
||||
headers=headers,
|
||||
)
|
||||
except ICloudPyAPIResponseException as error:
|
||||
if error.code == -21669:
|
||||
# Wrong verification code
|
||||
LOGGER.error("Code verification failed.")
|
||||
return False
|
||||
raise
|
||||
|
||||
LOGGER.debug("Code verification successful.")
|
||||
|
||||
self.trust_session()
|
||||
return not self.requires_2sa
|
||||
|
||||
def trust_session(self):
|
||||
"""Request session trust to avoid user log in going forward."""
|
||||
headers = self._get_auth_headers()
|
||||
|
||||
if self.session_data.get("scnt"):
|
||||
headers["scnt"] = self.session_data.get("scnt")
|
||||
|
||||
if self.session_data.get("session_id"):
|
||||
headers["X-Apple-ID-Session-Id"] = self.session_data.get("session_id")
|
||||
|
||||
try:
|
||||
self.session.get(
|
||||
f"{self.auth_endpoint}/2sv/trust",
|
||||
headers=headers,
|
||||
)
|
||||
self._authenticate_with_token()
|
||||
return True
|
||||
except ICloudPyAPIResponseException:
|
||||
LOGGER.error("Session trust failed.")
|
||||
return False
|
||||
|
||||
def _get_webservice_url(self, ws_key):
|
||||
"""Get webservice URL, raise an exception if not exists."""
|
||||
if self._webservices.get(ws_key) is None:
|
||||
raise ICloudPyServiceNotActivatedException(
|
||||
"Webservice not available", ws_key
|
||||
)
|
||||
return self._webservices[ws_key]["url"]
|
||||
|
||||
@property
|
||||
def devices(self):
|
||||
"""Returns all devices."""
|
||||
service_root = self._get_webservice_url("findme")
|
||||
return FindMyiPhoneServiceManager(
|
||||
service_root, self.session, self.params, self.with_family
|
||||
)
|
||||
|
||||
@property
|
||||
def iphone(self):
|
||||
"""Returns the iPhone."""
|
||||
return self.devices[0]
|
||||
|
||||
@property
|
||||
def drive(self):
|
||||
"""Gets the 'Drive' service."""
|
||||
if not self._drive:
|
||||
self._drive = DriveService(
|
||||
service_root=self._get_webservice_url("drivews"),
|
||||
document_root=self._get_webservice_url("docws"),
|
||||
session=self.session,
|
||||
params=self.params,
|
||||
)
|
||||
return self._drive
|
||||
|
||||
@property
|
||||
def work(self):
|
||||
"""Gets the 'Work' service."""
|
||||
if not self._work:
|
||||
self._work = WorkService(
|
||||
document_root=self._get_webservice_url("iworkexportws"),
|
||||
session=self.session,
|
||||
params=self.params,
|
||||
client_id=self.client_id,
|
||||
dsid=self.data["dsInfo"].get("dsid"),
|
||||
)
|
||||
return self._work
|
||||
|
||||
def __unicode__(self):
|
||||
return f"iCloud API: {self.user.get('accountName')}"
|
||||
|
||||
def __str__(self):
|
||||
as_unicode = self.__unicode__()
|
||||
if PY2:
|
||||
return as_unicode.encode("utf-8", "ignore")
|
||||
return as_unicode
|
||||
|
||||
def __repr__(self):
|
||||
return f"<{str(self)}>"
|
||||
@@ -0,0 +1,374 @@
|
||||
#! /usr/bin/env python
|
||||
"""
|
||||
A Command Line Wrapper to allow easy use of iCloudPy for
|
||||
command line scripts, and related.
|
||||
"""
|
||||
|
||||
# from builtins import input
|
||||
import argparse
|
||||
import pickle
|
||||
import sys
|
||||
|
||||
from click import confirm
|
||||
|
||||
from icloudpy import ICloudPyService, utils
|
||||
from icloudpy.exceptions import ICloudPyFailedLoginException
|
||||
|
||||
DEVICE_ERROR = "Please use the --device switch to indicate which device to use."
|
||||
|
||||
|
||||
def create_pickled_data(idevice, filename):
|
||||
"""
|
||||
This helper will output the idevice to a pickled file named
|
||||
after the passed filename.
|
||||
|
||||
This allows the data to be used without resorting to screen / pipe
|
||||
scrapping.
|
||||
"""
|
||||
pickle_file = open(filename, "wb")
|
||||
pickle.dump(idevice.content, pickle_file, protocol=pickle.HIGHEST_PROTOCOL)
|
||||
pickle_file.close()
|
||||
|
||||
|
||||
def main(args=None):
|
||||
"""Main commandline entrypoint."""
|
||||
if args is None:
|
||||
args = sys.argv[1:]
|
||||
|
||||
parser = argparse.ArgumentParser(description="Find My iPhone CommandLine Tool")
|
||||
|
||||
parser.add_argument(
|
||||
"--username",
|
||||
action="store",
|
||||
dest="username",
|
||||
default="",
|
||||
help="Apple ID to Use",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--password",
|
||||
action="store",
|
||||
dest="password",
|
||||
default="",
|
||||
help=(
|
||||
"Apple ID Password to Use; if unspecified, password will be "
|
||||
"fetched from the system keyring."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"-n",
|
||||
"--non-interactive",
|
||||
action="store_false",
|
||||
dest="interactive",
|
||||
default=True,
|
||||
help="Disable interactive prompts.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--delete-from-keyring",
|
||||
action="store_true",
|
||||
dest="delete_from_keyring",
|
||||
default=False,
|
||||
help="Delete stored password in system keyring for this username.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--list",
|
||||
action="store_true",
|
||||
dest="list",
|
||||
default=False,
|
||||
help="Short Listings for Device(s) associated with account",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--llist",
|
||||
action="store_true",
|
||||
dest="longlist",
|
||||
default=False,
|
||||
help="Detailed Listings for Device(s) associated with account",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--locate",
|
||||
action="store_true",
|
||||
dest="locate",
|
||||
default=False,
|
||||
help="Retrieve Location for the iDevice (non-exclusive).",
|
||||
)
|
||||
|
||||
# Restrict actions to a specific devices UID / DID
|
||||
parser.add_argument(
|
||||
"--device",
|
||||
action="store",
|
||||
dest="device_id",
|
||||
default=False,
|
||||
help="Only effect this device",
|
||||
)
|
||||
|
||||
# Trigger Sound Alert
|
||||
parser.add_argument(
|
||||
"--sound",
|
||||
action="store_true",
|
||||
dest="sound",
|
||||
default=False,
|
||||
help="Play a sound on the device",
|
||||
)
|
||||
|
||||
# Trigger Message w/Sound Alert
|
||||
parser.add_argument(
|
||||
"--message",
|
||||
action="store",
|
||||
dest="message",
|
||||
default=False,
|
||||
help="Optional Text Message to display with a sound",
|
||||
)
|
||||
|
||||
# Trigger Message (without Sound) Alert
|
||||
parser.add_argument(
|
||||
"--silentmessage",
|
||||
action="store",
|
||||
dest="silentmessage",
|
||||
default=False,
|
||||
help="Optional Text Message to display with no sounds",
|
||||
)
|
||||
|
||||
# Lost Mode
|
||||
parser.add_argument(
|
||||
"--lostmode",
|
||||
action="store_true",
|
||||
dest="lostmode",
|
||||
default=False,
|
||||
help="Enable Lost mode for the device",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--lostphone",
|
||||
action="store",
|
||||
dest="lost_phone",
|
||||
default=False,
|
||||
help="Phone Number allowed to call when lost mode is enabled",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--lostpassword",
|
||||
action="store",
|
||||
dest="lost_password",
|
||||
default=False,
|
||||
help="Forcibly active this passcode on the idevice",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--lostmessage",
|
||||
action="store",
|
||||
dest="lost_message",
|
||||
default="",
|
||||
help="Forcibly display this message when activating lost mode.",
|
||||
)
|
||||
|
||||
# Output device data to an pickle file
|
||||
parser.add_argument(
|
||||
"--outputfile",
|
||||
action="store_true",
|
||||
dest="output_to_file",
|
||||
default="",
|
||||
help="Save device data to a file in the current directory.",
|
||||
)
|
||||
|
||||
# Path to session directory
|
||||
parser.add_argument(
|
||||
"--session-directory",
|
||||
action="store",
|
||||
dest="session_directory",
|
||||
default=None,
|
||||
help="Path to save session information",
|
||||
)
|
||||
|
||||
# Server region - global or china
|
||||
parser.add_argument(
|
||||
"--region",
|
||||
action="store",
|
||||
dest="region",
|
||||
default="global",
|
||||
help="Server region - global or china",
|
||||
)
|
||||
|
||||
command_line = parser.parse_args(args)
|
||||
|
||||
username = command_line.username
|
||||
password = command_line.password
|
||||
session_directory = command_line.session_directory
|
||||
server_region = command_line.region
|
||||
|
||||
if username and command_line.delete_from_keyring:
|
||||
utils.delete_password_in_keyring(username)
|
||||
|
||||
failure_count = 0
|
||||
while True:
|
||||
# Which password we use is determined by your username, so we
|
||||
# do need to check for this first and separately.
|
||||
if not username:
|
||||
parser.error("No username supplied")
|
||||
|
||||
if not password:
|
||||
password = utils.get_password(
|
||||
username, interactive=command_line.interactive
|
||||
)
|
||||
|
||||
if not password:
|
||||
parser.error("No password supplied")
|
||||
|
||||
try:
|
||||
api = (
|
||||
ICloudPyService(
|
||||
apple_id=username.strip(),
|
||||
password=password.strip(),
|
||||
cookie_directory=session_directory,
|
||||
home_endpoint="https://www.icloud.com.cn",
|
||||
setup_endpoint="https://setup.icloud.com.cn/setup/ws/1",
|
||||
)
|
||||
if server_region == "china"
|
||||
else ICloudPyService(
|
||||
apple_id=username.strip(),
|
||||
password=password.strip(),
|
||||
cookie_directory=session_directory,
|
||||
)
|
||||
)
|
||||
|
||||
if (
|
||||
not utils.password_exists_in_keyring(username)
|
||||
and command_line.interactive
|
||||
and confirm("Save password in keyring?")
|
||||
):
|
||||
utils.store_password_in_keyring(username, password)
|
||||
|
||||
if api.requires_2fa:
|
||||
# fmt: off
|
||||
print(
|
||||
"\nTwo-step authentication required.",
|
||||
"\nPlease enter validation code"
|
||||
)
|
||||
# fmt: on
|
||||
|
||||
code = input("(string) --> ")
|
||||
if not api.validate_2fa_code(code):
|
||||
print("Failed to verify verification code")
|
||||
sys.exit(1)
|
||||
|
||||
print("")
|
||||
|
||||
elif api.requires_2sa:
|
||||
# fmt: off
|
||||
print(
|
||||
"\nTwo-step authentication required.",
|
||||
"\nYour trusted devices are:"
|
||||
)
|
||||
# fmt: on
|
||||
|
||||
devices = api.trusted_devices
|
||||
for i, device in enumerate(devices):
|
||||
print(
|
||||
f' {i}: {device.get("deviceName", "SMS to " + device.get("phoneNumber"))}'
|
||||
)
|
||||
|
||||
print("\nWhich device would you like to use?")
|
||||
device = int(input("(number) --> "))
|
||||
device = devices[device]
|
||||
if not api.send_verification_code(device):
|
||||
print("Failed to send verification code")
|
||||
sys.exit(1)
|
||||
|
||||
print("\nPlease enter validation code")
|
||||
code = input("(string) --> ")
|
||||
if not api.validate_verification_code(device, code):
|
||||
print("Failed to verify verification code")
|
||||
sys.exit(1)
|
||||
|
||||
print("")
|
||||
break
|
||||
except ICloudPyFailedLoginException as error:
|
||||
# If they have a stored password; we just used it and
|
||||
# it did not work; let's delete it if there is one.
|
||||
if utils.password_exists_in_keyring(username):
|
||||
utils.delete_password_in_keyring(username)
|
||||
|
||||
message = f"Bad username or password for {username}"
|
||||
password = None
|
||||
|
||||
failure_count += 1
|
||||
if failure_count >= 3:
|
||||
raise RuntimeError(message) from error
|
||||
|
||||
print(message, file=sys.stderr)
|
||||
|
||||
for dev in api.devices:
|
||||
if not command_line.device_id or (
|
||||
command_line.device_id.strip().lower() == dev.content["id"].strip().lower()
|
||||
):
|
||||
# List device(s)
|
||||
if command_line.locate:
|
||||
dev.location()
|
||||
|
||||
if command_line.output_to_file:
|
||||
create_pickled_data(
|
||||
dev,
|
||||
filename=(dev.content["name"].strip().lower() + ".fmip_snapshot"),
|
||||
)
|
||||
|
||||
contents = dev.content
|
||||
if command_line.longlist:
|
||||
print("-" * 30)
|
||||
print(contents["name"])
|
||||
for key in contents:
|
||||
print(f"{key} - {contents[key]}")
|
||||
elif command_line.list:
|
||||
print("-" * 30)
|
||||
print(f"Name - {contents['name']}")
|
||||
print(f"Display Name - {contents['deviceDisplayName']}")
|
||||
print(f"Location - {contents['location']}")
|
||||
print(f"Battery Level - {contents['batteryLevel']}")
|
||||
print(f"Battery Status- {contents['batteryStatus']}")
|
||||
print(f"Device Class - {contents['deviceClass']}")
|
||||
print(f"Device Model - {contents['deviceModel']}")
|
||||
|
||||
# Play a Sound on a device
|
||||
if command_line.sound:
|
||||
if command_line.device_id:
|
||||
dev.play_sound()
|
||||
else:
|
||||
raise RuntimeError(
|
||||
f"\n\n\t\tSounds can only be played on a singular device. {DEVICE_ERROR}\n\n"
|
||||
)
|
||||
|
||||
# Display a Message on the device
|
||||
if command_line.message:
|
||||
if command_line.device_id:
|
||||
dev.display_message(
|
||||
subject="A Message", message=command_line.message, sounds=True
|
||||
)
|
||||
else:
|
||||
raise RuntimeError(
|
||||
f"Messages can only be played on a singular device. {DEVICE_ERROR}"
|
||||
)
|
||||
|
||||
# Display a Silent Message on the device
|
||||
if command_line.silentmessage:
|
||||
if command_line.device_id:
|
||||
dev.display_message(
|
||||
subject="A Silent Message",
|
||||
message=command_line.silentmessage,
|
||||
sounds=False,
|
||||
)
|
||||
else:
|
||||
raise RuntimeError(
|
||||
f"Silent Messages can only be played on a singular device. {DEVICE_ERROR}"
|
||||
)
|
||||
|
||||
# Enable Lost mode
|
||||
if command_line.lostmode:
|
||||
if command_line.device_id:
|
||||
dev.lost_device(
|
||||
number=command_line.lost_phone.strip(),
|
||||
text=command_line.lost_message.strip(),
|
||||
newpasscode=command_line.lost_password.strip(),
|
||||
)
|
||||
else:
|
||||
raise RuntimeError(
|
||||
f"Lost Mode can only be activated on a singular device. {DEVICE_ERROR}"
|
||||
)
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,47 @@
|
||||
"""Library exceptions."""
|
||||
|
||||
|
||||
class ICloudPyException(Exception):
|
||||
"""Generic iCloud exception."""
|
||||
|
||||
|
||||
# API
|
||||
class ICloudPyAPIResponseException(ICloudPyException):
|
||||
"""iCloud response exception."""
|
||||
|
||||
def __init__(self, reason, code=None, retry=False):
|
||||
self.reason = reason
|
||||
self.code = code
|
||||
message = reason or ""
|
||||
if code:
|
||||
message += f" ({code})"
|
||||
if retry:
|
||||
message += ". Retrying ..."
|
||||
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
class ICloudPyServiceNotActivatedException(ICloudPyAPIResponseException):
|
||||
"""iCloud service not activated exception."""
|
||||
|
||||
|
||||
# Login
|
||||
class ICloudPyFailedLoginException(ICloudPyException):
|
||||
"""iCloud failed login exception."""
|
||||
|
||||
|
||||
class ICloudPy2SARequiredException(ICloudPyException):
|
||||
"""iCloud 2SA required exception."""
|
||||
|
||||
def __init__(self, apple_id):
|
||||
message = f"Two-step authentication required for account:{apple_id}"
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
class ICloudPyNoStoredPasswordAvailableException(ICloudPyException):
|
||||
"""iCloud no stored password exception."""
|
||||
|
||||
|
||||
# Webservice specific
|
||||
class ICloudPyNoDevicesException(ICloudPyException):
|
||||
"""iCloud no device exception."""
|
||||
@@ -0,0 +1,3 @@
|
||||
from icloudpy.services.drive import DriveService # pylint: disable=unused-import
|
||||
from icloudpy.services.work import WorkService # pylint: disable=unused-import
|
||||
|
||||
@@ -0,0 +1,385 @@
|
||||
"""Drive service."""
|
||||
import io
|
||||
import json
|
||||
import mimetypes
|
||||
import os
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from re import search
|
||||
|
||||
from requests import Response
|
||||
from six import PY2
|
||||
|
||||
|
||||
class DriveService:
|
||||
"""The 'Drive' iCloud service."""
|
||||
|
||||
def __init__(self, service_root, document_root, session, params):
|
||||
self._service_root = service_root
|
||||
self._document_root = document_root
|
||||
self.session = session
|
||||
self.params = dict(params)
|
||||
self._root = None
|
||||
|
||||
def _get_token_from_cookie(self):
|
||||
for cookie in self.session.cookies:
|
||||
if cookie.name == "X-APPLE-WEBAUTH-VALIDATE":
|
||||
match = search(r"\bt=([^:]+)", cookie.value)
|
||||
if match is None:
|
||||
raise Exception(f"Can't extract token from {cookie.value}")
|
||||
return {"token": match.group(1)}
|
||||
raise Exception("Token cookie not found")
|
||||
|
||||
def get_node_data(self, drivewsid):
|
||||
"""Returns the node data."""
|
||||
request = self.session.post(
|
||||
self._service_root + "/retrieveItemDetailsInFolders",
|
||||
params=self.params,
|
||||
data=json.dumps(
|
||||
[
|
||||
{
|
||||
"drivewsid": drivewsid,
|
||||
"partialData": False,
|
||||
}
|
||||
]
|
||||
),
|
||||
)
|
||||
if not request.ok:
|
||||
self.session.raise_error(request.status_code, request.reason)
|
||||
return request.json()[0]
|
||||
|
||||
def get_file(self, file_id, zone="com.apple.CloudDocs", **kwargs):
|
||||
"""Returns iCloud Drive file."""
|
||||
file_params = dict(self.params)
|
||||
file_params.update({"document_id": file_id})
|
||||
response = self.session.get(
|
||||
self._document_root + f"/ws/{zone}/download/by_id",
|
||||
params=file_params,
|
||||
)
|
||||
if not response.ok:
|
||||
self.session.raise_error(response.status_code, response.reason)
|
||||
package_token = response.json().get("package_token")
|
||||
data_token = response.json().get("data_token")
|
||||
if data_token and data_token.get("url"):
|
||||
return self.session.get(data_token["url"], params=self.params, **kwargs)
|
||||
elif package_token and package_token.get("url"):
|
||||
return self.session.get(package_token["url"], params=self.params, **kwargs)
|
||||
else:
|
||||
raise KeyError("'data_token' nor 'package_token' found in response.")
|
||||
|
||||
def get_app_data(self):
|
||||
"""Returns the app library (previously ubiquity)."""
|
||||
request = self.session.get(
|
||||
self._service_root + "/retrieveAppLibraries", params=self.params
|
||||
)
|
||||
if not request.ok:
|
||||
self.session.raise_error(request.status_code, request.reason)
|
||||
return request.json()["items"]
|
||||
|
||||
def get_app_node(self, app_id, folder="documents"):
|
||||
"""Returns the node of the app (ubiquity)"""
|
||||
return DriveNode(self, self.get_node_data("FOLDER::" + app_id + "::" + folder))
|
||||
|
||||
def _get_upload_contentws_url(self, file_object, zone="com.apple.CloudDocs"):
|
||||
"""Get the contentWS endpoint URL to add a new file."""
|
||||
content_type = mimetypes.guess_type(file_object.name)[0]
|
||||
if content_type is None:
|
||||
content_type = ""
|
||||
|
||||
# Get filesize from file object
|
||||
orig_pos = file_object.tell()
|
||||
file_object.seek(0, os.SEEK_END)
|
||||
file_size = file_object.tell()
|
||||
file_object.seek(orig_pos, os.SEEK_SET)
|
||||
|
||||
file_params = self.params
|
||||
file_params.update(self._get_token_from_cookie())
|
||||
|
||||
request = self.session.post(
|
||||
self._document_root + f"/ws/{zone}/upload/web",
|
||||
params=file_params,
|
||||
headers={"Content-Type": "text/plain"},
|
||||
data=json.dumps(
|
||||
{
|
||||
"filename": file_object.name,
|
||||
"type": "FILE",
|
||||
"content_type": content_type,
|
||||
"size": file_size,
|
||||
}
|
||||
),
|
||||
)
|
||||
if not request.ok:
|
||||
self.session.raise_error(request.status_code, request.reason)
|
||||
return (request.json()[0]["document_id"], request.json()[0]["url"])
|
||||
|
||||
def _update_contentws(
|
||||
self, folder_id, sf_info, document_id, file_object, zone="com.apple.CloudDocs"
|
||||
):
|
||||
data = {
|
||||
"data": {
|
||||
"signature": sf_info["fileChecksum"],
|
||||
"wrapping_key": sf_info["wrappingKey"],
|
||||
"reference_signature": sf_info["referenceChecksum"],
|
||||
"size": sf_info["size"],
|
||||
},
|
||||
"command": "add_file",
|
||||
"create_short_guid": True,
|
||||
"document_id": document_id,
|
||||
"path": {
|
||||
"starting_document_id": folder_id,
|
||||
"path": os.path.basename(file_object.name),
|
||||
},
|
||||
"allow_conflict": True,
|
||||
"file_flags": {
|
||||
"is_writable": True,
|
||||
"is_executable": False,
|
||||
"is_hidden": False,
|
||||
},
|
||||
"mtime": int(time.time() * 1000),
|
||||
"btime": int(time.time() * 1000),
|
||||
}
|
||||
|
||||
# Add the receipt if we have one. Will be absent for 0-sized files
|
||||
if sf_info.get("receipt"):
|
||||
data["data"].update({"receipt": sf_info["receipt"]})
|
||||
|
||||
request = self.session.post(
|
||||
self._document_root + f"/ws/{zone}/update/documents",
|
||||
params=self.params,
|
||||
headers={"Content-Type": "text/plain"},
|
||||
data=json.dumps(data),
|
||||
)
|
||||
if not request.ok:
|
||||
self.session.raise_error(request.status_code, request.reason)
|
||||
return request.json()
|
||||
|
||||
def send_file(self, folder_id, file_object, zone="com.apple.CloudDocs"):
|
||||
"""Send new file to iCloud Drive."""
|
||||
document_id, content_url = self._get_upload_contentws_url(file_object, zone)
|
||||
|
||||
request = self.session.post(content_url, files={file_object.name: file_object})
|
||||
if not request.ok:
|
||||
self.session.raise_error(request.status_code, request.reason)
|
||||
content_response = request.json()["singleFile"]
|
||||
|
||||
self._update_contentws(
|
||||
folder_id, content_response, document_id, file_object, zone
|
||||
)
|
||||
|
||||
def create_folders(self, parent, name):
|
||||
"""Creates a new iCloud Drive folder"""
|
||||
request = self.session.post(
|
||||
self._service_root + "/createFolders",
|
||||
params=self.params,
|
||||
headers={"Content-Type": "text/plain"},
|
||||
data=json.dumps(
|
||||
{
|
||||
"destinationDrivewsId": parent,
|
||||
"folders": [
|
||||
{
|
||||
"clientId": self.params["clientId"],
|
||||
"name": name,
|
||||
}
|
||||
],
|
||||
}
|
||||
),
|
||||
)
|
||||
return request.json()
|
||||
|
||||
def rename_items(self, node_id, etag, name):
|
||||
"""Renames an iCloud Drive node"""
|
||||
request = self.session.post(
|
||||
self._service_root + "/renameItems",
|
||||
params=self.params,
|
||||
data=json.dumps(
|
||||
{
|
||||
"items": [
|
||||
{
|
||||
"drivewsid": node_id,
|
||||
"etag": etag,
|
||||
"name": name,
|
||||
}
|
||||
],
|
||||
}
|
||||
),
|
||||
)
|
||||
return request.json()
|
||||
|
||||
def move_items_to_trash(self, node_id, etag):
|
||||
"""Moves an iCloud Drive node to the trash bin"""
|
||||
request = self.session.post(
|
||||
self._service_root + "/moveItemsToTrash",
|
||||
params=self.params,
|
||||
data=json.dumps(
|
||||
{
|
||||
"items": [
|
||||
{
|
||||
"drivewsid": node_id,
|
||||
"etag": etag,
|
||||
"clientId": self.params["clientId"],
|
||||
}
|
||||
],
|
||||
}
|
||||
),
|
||||
)
|
||||
if not request.ok:
|
||||
self.session.raise_error(request.status_code, request.reason)
|
||||
return request.json()
|
||||
|
||||
@property
|
||||
def root(self):
|
||||
"""Returns the root node."""
|
||||
if not self._root:
|
||||
self._root = DriveNode(
|
||||
self, self.get_node_data("FOLDER::com.apple.CloudDocs::root")
|
||||
)
|
||||
return self._root
|
||||
|
||||
def __getattr__(self, attr):
|
||||
return getattr(self.root, attr)
|
||||
|
||||
def __getitem__(self, key):
|
||||
return self.root[key]
|
||||
|
||||
|
||||
class DriveNode:
|
||||
"""Drive node."""
|
||||
|
||||
def __init__(self, conn, data):
|
||||
self.data = data
|
||||
self.connection = conn
|
||||
self._children = None
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Gets the node name."""
|
||||
if "extension" in self.data:
|
||||
return f'{self.data["name"]}.{self.data["extension"]}'
|
||||
return self.data["name"]
|
||||
|
||||
@property
|
||||
def type(self):
|
||||
"""Gets the node type."""
|
||||
node_type = self.data.get("type")
|
||||
return node_type and node_type.lower()
|
||||
|
||||
def get_children(self):
|
||||
"""Gets the node children."""
|
||||
if not self._children:
|
||||
if "items" not in self.data:
|
||||
self.data.update(self.connection.get_node_data(self.data["drivewsid"]))
|
||||
if "items" not in self.data:
|
||||
raise KeyError(f'No items in folder, status: {self.data["status"]}')
|
||||
self._children = [
|
||||
DriveNode(self.connection, item_data)
|
||||
for item_data in self.data["items"]
|
||||
]
|
||||
return self._children
|
||||
|
||||
@property
|
||||
def size(self):
|
||||
"""Gets the node size."""
|
||||
size = self.data.get("size") # Folder does not have size
|
||||
if not size:
|
||||
return None
|
||||
return int(size)
|
||||
|
||||
@property
|
||||
def date_created(self):
|
||||
"""Gets the node created date (in UTC)."""
|
||||
return _date_to_utc(self.data.get("dateCreated"))
|
||||
|
||||
@property
|
||||
def date_changed(self):
|
||||
"""Gets the node changed date (in UTC)."""
|
||||
return _date_to_utc(self.data.get("dateChanged")) # Folder does not have date
|
||||
|
||||
@property
|
||||
def date_modified(self):
|
||||
"""Gets the node modified date (in UTC)."""
|
||||
return _date_to_utc(self.data.get("dateModified")) # Folder does not have date
|
||||
|
||||
@property
|
||||
def date_last_open(self):
|
||||
"""Gets the node last open date (in UTC)."""
|
||||
return _date_to_utc(self.data.get("lastOpenTime")) # Folder does not have date
|
||||
|
||||
def open(self, **kwargs):
|
||||
"""Gets the node file."""
|
||||
# iCloud returns 400 Bad Request for 0-byte files
|
||||
if self.data["size"] == 0:
|
||||
response = Response()
|
||||
response.raw = io.BytesIO()
|
||||
return response
|
||||
return self.connection.get_file(
|
||||
self.data["docwsid"], zone=self.data["zone"], **kwargs
|
||||
)
|
||||
|
||||
def upload(self, file_object, **kwargs):
|
||||
""" "Upload a new file."""
|
||||
return self.connection.send_file(
|
||||
self.data["docwsid"], file_object, zone=self.data["zone"], **kwargs
|
||||
)
|
||||
|
||||
def dir(self):
|
||||
"""Gets the node list of directories."""
|
||||
if self.type == "file":
|
||||
return None
|
||||
return [child.name for child in self.get_children()]
|
||||
|
||||
def mkdir(self, folder):
|
||||
"""Create a new directory directory."""
|
||||
# remove cached entries information first so that it will be re-read on next get_children()
|
||||
self._children = None
|
||||
if "items" in self.data:
|
||||
self.data.pop("items")
|
||||
return self.connection.create_folders(self.data["drivewsid"], folder)
|
||||
|
||||
def rename(self, name):
|
||||
"""Rename an iCloud Drive item."""
|
||||
return self.connection.rename_items(
|
||||
self.data["drivewsid"], self.data["etag"], name
|
||||
)
|
||||
|
||||
def delete(self):
|
||||
"""Delete an iCloud Drive item."""
|
||||
return self.connection.move_items_to_trash(
|
||||
self.data["drivewsid"], self.data["etag"]
|
||||
)
|
||||
|
||||
def get(self, name):
|
||||
"""Gets the node child."""
|
||||
if self.type == "file":
|
||||
return None
|
||||
return [child for child in self.get_children() if child.name == name][0]
|
||||
|
||||
def __getitem__(self, key):
|
||||
try:
|
||||
return self.get(key)
|
||||
except IndexError as error:
|
||||
raise KeyError(f"No child named '{key}' exists") from error
|
||||
|
||||
def __unicode__(self):
|
||||
return f"{{type: {self.type}, name: {self.name}}}"
|
||||
|
||||
def __str__(self):
|
||||
as_unicode = self.__unicode__()
|
||||
if PY2:
|
||||
return as_unicode.encode("utf-8", "ignore")
|
||||
return as_unicode
|
||||
|
||||
def __repr__(self):
|
||||
return f"<{type(self).__name__}: {str(self)}>"
|
||||
|
||||
|
||||
def _date_to_utc(date):
|
||||
if not date:
|
||||
return None
|
||||
# jump through hoops to return time in UTC rather than California time
|
||||
match = search(r"^(.+?)([\+\-]\d+):(\d\d)$", date)
|
||||
if not match:
|
||||
# Already in UTC
|
||||
return datetime.strptime(date, "%Y-%m-%dT%H:%M:%SZ")
|
||||
base = datetime.strptime(match.group(1), "%Y-%m-%dT%H:%M:%S")
|
||||
diff = timedelta(hours=int(match.group(2)), minutes=int(match.group(3)))
|
||||
return base - diff
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Work service."""
|
||||
import io
|
||||
import json
|
||||
import mimetypes
|
||||
import os
|
||||
import time
|
||||
import shutil
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from re import search
|
||||
|
||||
from requests import Response
|
||||
from six import PY2
|
||||
from src import LOGGER, config_parser
|
||||
|
||||
class WorkService:
|
||||
|
||||
def __init__(self, document_root, session, params, client_id, dsid):
|
||||
|
||||
self._document_root = document_root
|
||||
self.session = session
|
||||
self.params = dict(params)
|
||||
self.client_id = client_id
|
||||
self.dsid = dsid
|
||||
self._root = None
|
||||
|
||||
def export_response(self, document_id, secret, zone):
|
||||
file_params = dict(self.params)
|
||||
file_params.update({"clientBuildNumber": "current"})
|
||||
file_params.update({"clientMasteringNumber": "Mcurrent"})
|
||||
file_params.update({"dsid": self.dsid})
|
||||
request = self.session.post(
|
||||
self._document_root + f"/iw/export-ws/{self.dsid}/export_document",
|
||||
params=file_params,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
data={
|
||||
"primary": "primary",
|
||||
"document_type": "numbers",
|
||||
"format": "org.openxmlformats.spreadsheetml.sheet",
|
||||
"locale": "en",
|
||||
"encrypt_result": "N",
|
||||
"secret": secret,
|
||||
"document_id": document_id,
|
||||
"zone": zone,
|
||||
}
|
||||
,
|
||||
)
|
||||
if not request.ok:
|
||||
self.session.raise_error(request.status_code, request.reason)
|
||||
|
||||
return request.json()["job_id"]
|
||||
|
||||
def check_job(self, job_id):
|
||||
url = self._document_root + f"/iw/export-ws/{self.dsid}/check_export_status"
|
||||
file_params = dict(self.params)
|
||||
file_params.update({"build": "primary"})
|
||||
file_params.update({"clientBuildNumber": "current"})
|
||||
file_params.update({"clientMasteringNumber": "Mcurrent"})
|
||||
file_params.update({"job_id": job_id})
|
||||
file_params.update({"dsid": self.dsid})
|
||||
request = self.session.post(
|
||||
url,
|
||||
params=file_params,
|
||||
headers={"Content-Type": "text/plain"},
|
||||
data={},
|
||||
)
|
||||
if not request.ok:
|
||||
self.session.raise_error(request.status_code, request.reason)
|
||||
if request.json()["job_status"] == "failure":
|
||||
raise Exception(request.json()["job_status"])
|
||||
if request.json()["job_status"] == "success":
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def download_file(self, job_id, destination_path, name):
|
||||
url = self._document_root + f"/iw/export-ws/{self.dsid}/download_exported_document"
|
||||
file_params = dict(self.params)
|
||||
file_params.update({"build": "primary"})
|
||||
file_params.update({"file_name": name + f".xlsx"})
|
||||
file_params.update({"job_id": job_id})
|
||||
local_filename = os.path.join(destination_path, name + f".xlsx")
|
||||
try:
|
||||
response = self.session.get(url, params=file_params, stream=True)
|
||||
with open(local_filename, 'wb') as out_file:
|
||||
shutil.copyfileobj(response.raw, out_file)
|
||||
except self.session.exceptions.RequestException as e:
|
||||
raise Exception(f"Ошибка скачивания сконцентрированного файла {local_filename}: {str(e)}")
|
||||
return local_filename
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
"""Utils."""
|
||||
import getpass
|
||||
from sys import stdout
|
||||
|
||||
import keyring
|
||||
|
||||
from .exceptions import ICloudPyNoStoredPasswordAvailableException
|
||||
|
||||
KEYRING_SYSTEM = "icloudpy://icloud-password"
|
||||
|
||||
|
||||
def get_password(username, interactive=stdout.isatty() if stdout else False):
|
||||
"""Get the password from a username."""
|
||||
try:
|
||||
return get_password_from_keyring(username)
|
||||
except ICloudPyNoStoredPasswordAvailableException:
|
||||
if not interactive:
|
||||
raise
|
||||
|
||||
return getpass.getpass(f"Enter iCloud password for {username}: ")
|
||||
|
||||
|
||||
def password_exists_in_keyring(username):
|
||||
"""Return true if the password of a username exists in the keyring."""
|
||||
try:
|
||||
get_password_from_keyring(username)
|
||||
except ICloudPyNoStoredPasswordAvailableException:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def get_password_from_keyring(username):
|
||||
"""Get the password from a username."""
|
||||
result = keyring.get_password(KEYRING_SYSTEM, username)
|
||||
if result is None:
|
||||
raise ICloudPyNoStoredPasswordAvailableException(
|
||||
f"No iCloudPy password for {username} could be found "
|
||||
"in the system keychain. Use the `--store-in-keyring` "
|
||||
"command-line option for storing a password for this "
|
||||
"username."
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def store_password_in_keyring(username, password):
|
||||
"""Store the password of a username."""
|
||||
return keyring.set_password(KEYRING_SYSTEM, username, password)
|
||||
|
||||
|
||||
def delete_password_in_keyring(username):
|
||||
"""Delete the password of a username."""
|
||||
return keyring.delete_password(KEYRING_SYSTEM, username)
|
||||
|
||||
|
||||
def underscore_to_camelcase(word, initial_capital=False):
|
||||
"""Transform a word to camelCase."""
|
||||
words = [x.capitalize() or "_" for x in word.split("_")]
|
||||
if not initial_capital:
|
||||
words[0] = words[0].lower()
|
||||
|
||||
return "".join(words)
|
||||
Reference in New Issue
Block a user