Implement two-factor authentication

Closes #200.

docs/as-module.rst

@@ -188,6 +188,10 @@ Exceptions
 
 .. autoexception:: LoginRequiredException
 
+.. autoexception:: TwoFactorAuthRequiredException
+
+   .. versionadded:: 4.2
+
 .. autoexception:: InvalidArgumentException
 
 .. autoexception:: BadResponseException

instaloader/__main__.py

@@ -8,7 +8,8 @@ from argparse import ArgumentParser, SUPPRESS
 from typing import List, Optional
 
 from . import (Instaloader, InstaloaderException, InvalidArgumentException, Post, Profile, ProfileNotExistsException,
-               StoryItem, __version__, load_structure_from_file)
+               StoryItem, __version__, load_structure_from_file, TwoFactorAuthRequiredException,
+               BadCredentialsException)
 from .instaloader import get_default_session_filename
 from .instaloadercontext import default_user_agent
 
@@ -84,7 +85,16 @@ def _main(instaloader: Instaloader, targetlist: List[str],
             instaloader.context.log("Session file does not exist yet - Logging in.")
         if not instaloader.context.is_logged_in or username != instaloader.test_login():
             if password is not None:
-                instaloader.login(username, password)
+                try:
+                    instaloader.login(username, password)
+                except TwoFactorAuthRequiredException:
+                    while True:
+                        try:
+                            code = input("Enter 2FA verification code: ")
+                            instaloader.two_factor_login(code)
+                            break
+                        except BadCredentialsException:
+                            pass
             else:
                 instaloader.interactive_login(username)
         instaloader.context.log("Logged in as %s." % username)

instaloader/exceptions.py

@@ -33,6 +33,10 @@ class LoginRequiredException(InstaloaderException):
     pass
 
 
+class TwoFactorAuthRequiredException(InstaloaderException):
+    pass
+
+
 class InvalidArgumentException(InstaloaderException):
     pass
 

instaloader/instaloader.py

@@ -350,9 +350,20 @@ class Instaloader:
 
         :raises InvalidArgumentException: If the provided username does not exist.
         :raises BadCredentialsException: If the provided password is wrong.
-        :raises ConnectionException: If connection to Instagram failed."""
+        :raises ConnectionException: If connection to Instagram failed.
+        :raises TwoFactorAuthRequiredException: First step of 2FA login done, now call :meth:`Instaloader.two_factor_login`."""
         self.context.login(user, passwd)
 
+    def two_factor_login(self, two_factor_code) -> None:
+        """Second step of login if 2FA is enabled.
+        Not meant to be used directly, use :meth:`Instaloader.two_factor_login`.
+
+        :raises InvalidArgumentException: No two-factor authentication pending.
+        :raises BadCredentialsException: 2FA verification code invalid.
+
+        .. versionadded:: 4.2"""
+        self.context.two_factor_login(two_factor_code)
+
     def format_filename(self, item: Union[Post, StoryItem], target: Optional[str] = None):
         """Format filename of a :class:`Post` or :class:`StoryItem` according to ``filename-pattern`` parameter.
 
@@ -1041,11 +1052,20 @@ class Instaloader:
         :raises ConnectionException: If connection to Instagram failed."""
         if self.context.quiet:
             raise LoginRequiredException("Quiet mode requires given password or valid session file.")
-        password = None
+        try:
-        while password is None:
+            password = None
-            password = getpass.getpass(prompt="Enter Instagram password for %s: " % username)
+            while password is None:
-            try:
+                password = getpass.getpass(prompt="Enter Instagram password for %s: " % username)
-                self.login(username, password)
+                try:
-            except BadCredentialsException as err:
+                    self.login(username, password)
-                print(err, file=sys.stderr)
+                except BadCredentialsException as err:
-                password = None
+                    print(err, file=sys.stderr)
+                    password = None
+        except TwoFactorAuthRequiredException:
+            while True:
+                try:
+                    code = input("Enter 2FA verification code: ")
+                    self.two_factor_login(code)
+                    break
+                except BadCredentialsException:
+                    pass

instaloader/instaloadercontext.py

@@ -59,6 +59,7 @@ class InstaloaderContext:
         self._graphql_page_length = 50
         self.graphql_count_per_slidingwindow = graphql_count_per_slidingwindow or 200
         self._root_rhx_gis = None
+        self.two_factor_auth_pending = None
 
         # error log, filled with error() and printed at the end of Instaloader.main()
         self.error_log = []
@@ -180,25 +181,32 @@ class InstaloaderContext:
 
         :raises InvalidArgumentException: If the provided username does not exist.
         :raises BadCredentialsException: If the provided password is wrong.
-        :raises ConnectionException: If connection to Instagram failed."""
+        :raises ConnectionException: If connection to Instagram failed.
+        :raises TwoFactorAuthRequiredException: First step of 2FA login done, now call :meth:`Instaloader.two_factor_login`."""
         import http.client
         # pylint:disable=protected-access
         http.client._MAXHEADERS = 200
         session = requests.Session()
         session.cookies.update({'sessionid': '', 'mid': '', 'ig_pr': '1',
-                                'ig_vw': '1920', 'csrftoken': '',
+                                'ig_vw': '1920', 'ig_cb': '1', 'csrftoken': '',
                                 's_network': '', 'ds_user_id': ''})
         session.headers.update(self._default_http_header())
-        session.headers.update({'X-CSRFToken': self.get_json('', {})['config']['csrf_token']})
+        session.get('https://www.instagram.com/web/__mid/')
+        csrf_token = session.cookies.get_dict()['csrftoken']
+        session.headers.update({'X-CSRFToken': csrf_token})
         # Not using self.get_json() here, because we need to access csrftoken cookie
         self._sleep()
         login = session.post('https://www.instagram.com/accounts/login/ajax/',
                              data={'password': passwd, 'username': user}, allow_redirects=True)
-        if login.status_code != 200:
-            if login.status_code == 400 and login.json().get('two_factor_required', None):
-                raise ConnectionException("Login error: Two factor authorization not yet supported.")
-            raise ConnectionException("Login error: {} {}".format(login.status_code, login.reason))
         resp_json = login.json()
+        if resp_json.get('two_factor_required'):
+            two_factor_session = copy_session(session)
+            two_factor_session.headers.update({'X-CSRFToken': csrf_token})
+            two_factor_session.cookies.update({'csrftoken': csrf_token})
+            self.two_factor_auth_pending = (two_factor_session,
+                                            user,
+                                            resp_json['two_factor_info']['two_factor_identifier'])
+            raise TwoFactorAuthRequiredException("Login error: two-factor authentication required.")
         if resp_json['status'] != 'ok':
             if 'message' in resp_json:
                 raise ConnectionException("Login error: \"{}\" status, message \"{}\".".format(resp_json['status'],
@@ -220,6 +228,32 @@ class InstaloaderContext:
         self._session = session
         self.username = user
 
+    def two_factor_login(self, two_factor_code):
+        """Second step of login if 2FA is enabled.
+        Not meant to be used directly, use :meth:`Instaloader.two_factor_login`.
+
+        :raises InvalidArgumentException: No two-factor authentication pending.
+        :raises BadCredentialsException: 2FA verification code invalid.
+
+        .. versionadded:: 4.2"""
+        if not self.two_factor_auth_pending:
+            raise InvalidArgumentException("No two-factor authentication pending.")
+        (session, user, two_factor_id) = self.two_factor_auth_pending
+
+        login = session.post('https://www.instagram.com/accounts/login/ajax/two_factor/',
+                             data={'username': user, 'verificationCode': two_factor_code, 'identifier': two_factor_id},
+                             allow_redirects=True)
+        resp_json = login.json()
+        if resp_json['status'] != 'ok':
+            if 'message' in resp_json:
+                raise BadCredentialsException("Login error: {}".format(resp_json['message']))
+            else:
+                raise BadCredentialsException("Login error: \"{}\" status.".format(resp_json['status']))
+        session.headers.update({'X-CSRFToken': login.cookies['csrftoken']})
+        self._session = session
+        self.username = user
+        self.two_factor_auth_pending = None
+
     def _sleep(self):
         """Sleep a short time if self.sleep is set. Called before each request to instagram.com."""
         if self.sleep: