Added support for LastPass Authenticator and trusting endpoint

Author: Terry Hardie

lastpass/exceptions.py

@@ -49,6 +49,11 @@ class LastPassIncorrectYubikeyPasswordError(Error):
     pass
 
 
+class LastPassIncorrectOutOfBandRequiredError(Error):
+    """LastPass error: need to provide out of band authentication (e.g, LastPass Authenticator)"""
+    pass
+
+
 class LastPassUnknownError(Error):
     """LastPass error we don't know about"""
     pass

lastpass/fetcher.py

@@ -1,5 +1,7 @@
 # coding: utf-8
 import hashlib
+import random
+import string
 from base64 import b64decode
 from binascii import hexlify
 import requests
@@ -13,6 +15,7 @@
     LastPassInvalidPasswordError,
     LastPassIncorrectGoogleAuthenticatorCodeError,
     LastPassIncorrectYubikeyPasswordError,
+    LastPassIncorrectOutOfBandRequiredError,
     LastPassUnknownError
 )
 from .session import Session
@@ -21,9 +24,9 @@
 http = requests
 
 
-def login(username, password, multifactor_password=None, client_id=None):
+def login(username, password, multifactor_password=None, client_id=None, trust_id=None, trust_me=False):
     key_iteration_count = request_iteration_count(username)
-    return request_login(username, password, key_iteration_count, multifactor_password, client_id)
+    return request_login(username, password, key_iteration_count, multifactor_password, client_id, trust_id=trust_id, trust_me=trust_me)
 
 
 def logout(session, web_client=http):
@@ -63,21 +66,31 @@ def request_iteration_count(username, web_client=http):
     raise InvalidResponseError('Key iteration count is not positive')
 
 
-def request_login(username, password, key_iteration_count, multifactor_password=None, client_id=None, web_client=http):
+def request_login(username, password, key_iteration_count, multifactor_password=None, client_id=None, web_client=http, trust_id=None, trust_me=False):
     body = {
-        'method': 'mobile',
-        'web': 1,
-        'xml': 1,
+        'method': 'cli',
+        'xml': 2,
         'username': username,
         'hash': make_hash(username, password, key_iteration_count),
         'iterations': key_iteration_count,
+        'includeprivatekeyenc': 1,
+        'outofbandsupported': 1
     }
 
     if multifactor_password:
         body['otp'] = multifactor_password
 
+    # if client_id:
+    #     body['imei'] = client_id
+
+    if trust_me and not trust_id:
+        trust_id = generate_trust_id()
+
+    if trust_id:
+        body['uuid'] = trust_id
+
     if client_id:
-        body['imei'] = client_id
+        body['trustlabel'] = client_id
 
     response = web_client.post('https://lastpass.com/login.php',
                                data=body)
@@ -93,17 +106,84 @@ def request_login(username, password, key_iteration_count, multifactor_password=
     if parsed_response is None:
         raise InvalidResponseError()
 
-    session = create_session(parsed_response, key_iteration_count)
+    session = create_session(parsed_response, key_iteration_count, trust_id)
     if not session:
-        raise login_error(parsed_response)
+        try:
+            raise login_error(parsed_response)
+        except LastPassIncorrectOutOfBandRequiredError:
+            (session, parsed_response) = oob_login(web_client, parsed_response,
+                                                   body, key_iteration_count,
+                                                   trust_id)
+            if not session:
+                raise login_error(parsed_response)
+    if trust_me:
+        response = web_client.post('https://lastpass.com/trust.php',
+                                   cookies={'PHPSESSID': session.id},
+                                   data={
+                                       "token": session.token,
+                                       "uuid": trust_id,
+                                       "trustlabel": client_id,
+                                   })
+
     return session
 
 
-def create_session(parsed_response, key_iteration_count):
+    if trust_me:
+        body['trustlabel'] = client_id
+
+
+def oob_login(web_client, parsed_response, body, key_iteration_count, trust_id):
+    error = None if parsed_response.tag != 'response' else parsed_response.find(
+        'error')
+    if 'outofbandname' not in error.attrib or 'capabilities' not in error.attrib:
+        return (None, parsed_response)
+    oob_name = error.attrib['outofbandname']
+    oob_capabilities = error.attrib['capabilities'].split(',')
+    can_do_passcode = 'passcode' in oob_capabilities
+    if can_do_passcode and 'outofband' not in oob_capabilities:
+        return (None, parsed_response)
+    body['outofbandrequest'] = '1'
+    # loop waiting for out of band approval, or failure
+    while True:
+        response = web_client.post("https://lastpass.com/login.php", data=body)
+        if response.status_code != requests.codes.ok:
+            raise NetworkError()
+
+        try:
+            parsed_response = etree.fromstring(response.content)
+        except etree.ParseError:
+            parsed_response = None
+
+        if parsed_response is None:
+            raise InvalidResponseError()
+
+        session = create_session(parsed_response, key_iteration_count, trust_id)
+        if session:
+            return (session, parsed_response)
+        error = None if parsed_response.tag != 'response' else parsed_response.find(
+            'error')
+        if 'cause' in error.attrib and error.attrib['cause'] == 'outofbandrequired':
+            if 'retryid' in error.attrib:
+                body['outofbandretryid'] = error.attrib['retryid']
+            body['outofbandretry'] = "1"
+            continue
+        return (None, parsed_response)
+
+
+def generate_trust_id():
+    return ''.join(random.choice(string.ascii_uppercase + string.digits + string.ascii_lowercase + "!@#$") for _ in range(32))
+
+
+def create_session(parsed_response, key_iteration_count, trust_id):
     if parsed_response.tag == 'ok':
-        session_id = parsed_response.attrib.get('sessionid')
+        ok_response = parsed_response
+    else:
+        ok_response = parsed_response.find("ok")
+    if ok_response is not None:
+        session_id = ok_response.attrib.get('sessionid')
+        token = ok_response.attrib.get('token')
         if isinstance(session_id, str):
-            return Session(session_id, key_iteration_count)
+            return Session(session_id, key_iteration_count, token, trust_id)
 
 
 def login_error(parsed_response):
@@ -117,6 +197,7 @@ def login_error(parsed_response):
         "googleauthrequired": LastPassIncorrectGoogleAuthenticatorCodeError,
         "googleauthfailed": LastPassIncorrectGoogleAuthenticatorCodeError,
         "yubikeyrestricted": LastPassIncorrectYubikeyPasswordError,
+        "outofbandrequired": LastPassIncorrectOutOfBandRequiredError,
     }
 
     cause = error.attrib.get('cause')

lastpass/session.py

@@ -1,8 +1,12 @@
 # coding: utf-8
 class Session(object):
-    def __init__(self, id, key_iteration_count):
+    def __init__(self, id, key_iteration_count, token=None, trust_id=None):
         self.id = id
         self.key_iteration_count = key_iteration_count
+        self.token = token
+        self.trust_id = trust_id
 
     def __eq__(self, other):
-        return self.id == other.id and self.key_iteration_count == other.key_iteration_count
+        return self.id == other.id and \
+               self.key_iteration_count == other.key_iteration_count and \
+               self.token == other.token

lastpass/vault.py

@@ -6,10 +6,10 @@
 
 class Vault(object):
     @classmethod
-    def open_remote(cls, username, password, multifactor_password=None, client_id=None):
+    def open_remote(cls, username, password, multifactor_password=None, client_id=None, trust_id=None, trust_me=False):
         """Fetches a blob from the server and creates a vault"""
-        blob = cls.fetch_blob(username, password, multifactor_password, client_id)
-        return cls.open(blob, username, password)
+        (blob, trust_id) = cls.fetch_blob(username, password, multifactor_password, client_id, trust_id, trust_me)
+        return cls.open(blob, username, password, trust_id)
 
     @classmethod
     def open_local(cls, blob_filename, username, password):
@@ -18,27 +18,28 @@ def open_local(cls, blob_filename, username, password):
         raise NotImplementedError()
 
     @classmethod
-    def open(cls, blob, username, password):
+    def open(cls, blob, username, password, trust_id=None):
         """Creates a vault from a blob object"""
-        return cls(blob, blob.encryption_key(username, password))
+        return cls(blob, blob.encryption_key(username, password), trust_id)
 
     @classmethod
-    def fetch_blob(cls, username, password, multifactor_password=None, client_id=None):
+    def fetch_blob(cls, username, password, multifactor_password=None, client_id=None, trust_id=None, trust_me=False):
         """Just fetches the blob, could be used to store it locally"""
-        session = fetcher.login(username, password, multifactor_password, client_id)
+        session = fetcher.login(username, password, multifactor_password, client_id, trust_id=trust_id, trust_me=trust_me)
         blob = fetcher.fetch(session)
         fetcher.logout(session)
 
-        return blob
+        return (blob, session.trust_id)
 
-    def __init__(self, blob, encryption_key):
+    def __init__(self, blob, encryption_key, trust_id=None):
         """This more of an internal method, use one of the static constructors instead"""
         chunks = parser.extract_chunks(blob)
 
         if not self.is_complete(chunks):
             raise InvalidResponseError('Blob is truncated')
 
         self.accounts = self.parse_accounts(chunks, encryption_key)
+        self.trust_id = trust_id
 
     def is_complete(self, chunks):
         return len(chunks) > 0 and chunks[-1].id == b'ENDM' and chunks[-1].payload == b'OK'

tests/test_fetcher.py

@@ -22,16 +22,17 @@ def setUp(self):
         self.blob_bytes = b64decode(self.blob_response)
         self.blob = Blob(self.blob_bytes, self.key_iteration_count)
 
-        self.login_post_data = {'method': 'mobile',
-                                'web': 1,
-                                'xml': 1,
+        self.login_post_data = {'method': 'cli',
+                                'xml': 2,
+                                'outofbandsupported': 1,
+                                'includeprivatekeyenc': 1,
                                 'username': self.username,
                                 'hash': self.hash,
                                 'iterations': self.key_iteration_count}
 
         self.device_id = '492378378052455'
         self.login_post_data_with_device_id = self.login_post_data.copy()
-        self.login_post_data_with_device_id.update({'imei': self.device_id})
+        self.login_post_data_with_device_id.update({'trustlabel': self.device_id})
 
         self.google_authenticator_code = '12345'
         self.yubikey_password = 'emdbwzemyisymdnevznyqhqnklaqheaxszzvtnxjrmkb'
@@ -162,7 +163,8 @@ def test_fetch_makes_a_get_request(self):
     def test_fetch_returns_a_blob(self):
         m = mock.Mock()
         m.get.return_value = self._http_ok(self.blob_response)
-        self.assertEqual(fetcher.fetch(self.session, m), self.blob)
+        returned_blob = fetcher.fetch(self.session, m)
+        self.assertEqual(returned_blob, self.blob)
 
     def test_fetch_raises_exception_on_http_error(self):
         m = mock.Mock()