From 4ee867c61b41db14cef7bfb7897519046540a242 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Andr=C3=A9=20Koch-Kramer?= <koch-kramer@web.de>
Date: Fri, 21 Dec 2018 21:52:17 +0100
Subject: [PATCH] Implement two-factor authentication
Closes #200.
---
docs/as-module.rst | 4 +++
instaloader/__main__.py | 14 +++++++--
instaloader/exceptions.py | 4 +++
instaloader/instaloader.py | 38 ++++++++++++++++++------
instaloader/instaloadercontext.py | 48 ++++++++++++++++++++++++++-----
5 files changed, 90 insertions(+), 18 deletions(-)
diff --git a/docs/as-module.rst b/docs/as-module.rst
index 44a8cce..79c545c 100644
--- a/docs/as-module.rst
+++ b/docs/as-module.rst
@@ -188,6 +188,10 @@ Exceptions
.. autoexception:: LoginRequiredException
+.. autoexception:: TwoFactorAuthRequiredException
+
+ .. versionadded:: 4.2
+
.. autoexception:: InvalidArgumentException
.. autoexception:: BadResponseException
diff --git a/instaloader/__main__.py b/instaloader/__main__.py
index a387fe9..9cd2a49 100644
--- a/instaloader/__main__.py
+++ b/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)
diff --git a/instaloader/exceptions.py b/instaloader/exceptions.py
index 287a3a8..29b55a9 100644
--- a/instaloader/exceptions.py
+++ b/instaloader/exceptions.py
@@ -33,6 +33,10 @@ class LoginRequiredException(InstaloaderException):
pass
+class TwoFactorAuthRequiredException(InstaloaderException):
+ pass
+
+
class InvalidArgumentException(InstaloaderException):
pass
diff --git a/instaloader/instaloader.py b/instaloader/instaloader.py
index 8728232..c43a1d0 100644
--- a/instaloader/instaloader.py
+++ b/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
- while password is None:
- password = getpass.getpass(prompt="Enter Instagram password for %s: " % username)
- try:
- self.login(username, password)
- except BadCredentialsException as err:
- print(err, file=sys.stderr)
- password = None
+ try:
+ password = None
+ while password is None:
+ password = getpass.getpass(prompt="Enter Instagram password for %s: " % username)
+ try:
+ self.login(username, password)
+ except BadCredentialsException as err:
+ 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
diff --git a/instaloader/instaloadercontext.py b/instaloader/instaloadercontext.py
index 425b39a..2d605c2 100644
--- a/instaloader/instaloadercontext.py
+++ b/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: