ValvePython.steam/steam/guard.py

"""
This submodule contains various functionality related to Steam Guard.

:class:`SteamAuthenticator` provides methods for genereating codes
and enabling 2FA on a Steam account. Operations managing the authenticator
on an account require an instance of either :class:`.MobileWebAuth` or
:class:`.SteamClient`. The instance needs to be logged in.

Adding an authenticator

.. code:: python

    wa = MobileWebAuth('steamuser')
    wa.cli_login()

    sa = SteamAuthenticator(backend=wa)
    sa.add()    # SMS code will be send to the account's phone number
    sa.secrets  # dict with authenticator secrets (SAVE THEM!!)

    # save the secrets, for example to a file
    json.dump(sa.secrets, open('./mysecrets.json', 'w'))

    # HINT: You can stop here and add authenticator on your phone.
    #       The secrets will be the same, and you will be able to
    #       both use your phone and SteamAuthenticator.

    sa.finalize('SMS CODE')  # activate the authenticator
    sa.get_code()  # generate 2FA code for login
    sa.remove()  # removes the authenticator from the account

.. warning::
    Before you finalize the authenticator, make sure to save your secrets.
    Otherwise you will lose access to the account.

Once authenticator is enabled all you need is the secrets to generate codes.

.. code:: python

    secrets = json.load(open('./mysecrets.json'))

    sa = SteamAuthenticator(secrets)
    sa.get_code()

You can obtain the authenticator secrets from an Android device using
:func:`extract_secrets_from_android_rooted`. See the function docstring for
details on what is required for it to work.

Format of ``secrets.json`` file:

.. code:: json

    {
        "account_name": "<username>",               # account username
        "identity_secret": "<base64 encoded>",      # unknown
        "revocation_code": "R51234",                # revocation code
        "secret_1": "<base54 encoded>",             # unknown
        "serial_number": "1111222333344445555",     # serial number
        "server_time": "1600000000",                # creation timestamp
        "shared_secret": "<base65 encoded>",        # secret used for code generation
        "status": 1,                                # status, 1 = token active
        "token_gid": "a1a1a1a1a1a1a1a1",            # token gid
        "uri": "otpauth://totp/Steam:<username>?secret=<base32 encoded shared seceret>&issuer=Steam"
    }
"""
import json
import subprocess
import struct
import requests
from base64 import b64decode, b64encode
from binascii import hexlify
from time import time
from steam import webapi
from steam.enums import ETwoFactorTokenType
from steam.steamid import SteamID
from steam.core.crypto import hmac_sha1, sha1_hash
from steam.enums.common import EResult
from steam.webauth import MobileWebAuth
from steam.utils.proto import proto_to_dict


class SteamAuthenticator(object):
    """Add/Remove authenticator from an account. Generate 2FA and confirmation codes."""
    _finalize_attempts = 5
    backend = None               #: instance of :class:`.MobileWebAuth` or :class:`.SteamClient`
    steam_time_offset = None     #: offset from steam server time
    align_time_every = 0         #: how often to align time with Steam (``0`` never, otherwise interval in seconds)
    _offset_last_check = 0
    secrets = None               #: :class:`dict` with authenticator secrets

    def __init__(self, secrets=None, backend=None):
        """
        :param secret: a dict of authenticator secrets
        :type  secret: dict
        :param backend: logged on session for steam user
        :type  backend: :class:`.MobileWebAuth`, :class:`.SteamClient`
        """
        self.secrets = secrets or {}
        self.backend = backend

    def __getattr__(self, key):
        if key not in self.secrets:
            raise AttributeError("No %s attribute" % repr(key))
        return self.secrets[key]

    def get_time(self):
        """
        :return: Steam aligned timestamp
        :rtype: int
        """
        if (self.steam_time_offset is None
           or (self.align_time_every and (time() - self._offset_last_check) > self.align_time_every)
           ):
            self.steam_time_offset = get_time_offset()

            if self.steam_time_offset is not None:
                self._offset_last_check = time()

        return int(time() + (self.steam_time_offset or 0))

    def get_code(self, timestamp=None):
        """
        :param timestamp: time to use for code generation
        :type timestamp: int
        :return: two factor code
        :rtype: str
        """
        return generate_twofactor_code_for_time(b64decode(self.shared_secret),
                                                self.get_time() if timestamp is None else timestamp)

    def get_confirmation_key(self, tag='', timestamp=None):
        """
        :param tag: see :func:`generate_confirmation_key` for this value
        :type tag: str
        :param timestamp: time to use for code generation
        :type timestamp: int
        :return: trade confirmation key
        :rtype: str
        """
        return generate_confirmation_key(b64decode(self.identity_secret), tag,
                                         self.get_time() if timestamp is None else timestamp)

    def _send_request(self, action, params):
        backend = self.backend

        if isinstance(backend, MobileWebAuth):
            if not backend.logged_on:
                raise SteamAuthenticatorError("MobileWebAuth instance not logged in")

            params['access_token'] = backend.oauth_token
            params['http_timeout'] = 10

            try:
                resp = webapi.post('ITwoFactorService', action, 1, params=params)
            except requests.exceptions.RequestException as exp:
                raise SteamAuthenticatorError("Error adding via WebAPI: %s" % str(exp))

            resp = resp['response']
        else:
            if not backend.logged_on:
                raise SteamAuthenticatorError("SteamClient instance not logged in")

            resp = backend.send_um_and_wait("TwoFactor.%s#1" % action,
                                            params, timeout=10)

            if resp is None:
                raise SteamAuthenticatorError("Failed. Request timeout")
            if resp.header.eresult != EResult.OK:
                raise SteamAuthenticatorError("Failed: %s (%s)" % (resp.header.error_message,
                                                                   repr(resp.header.eresult)))

            resp = proto_to_dict(resp.body)

            if action == 'AddAuthenticator':
                for key in ['shared_secret', 'identity_secret', 'secret_1']:
                    resp[key] = b64encode(resp[key]).decode('ascii')

        return resp

    def add(self):
        """Add authenticator to an account.
        The account's phone number will receive a SMS code required for :meth:`finalize`.

        :raises: :class:`SteamAuthenticatorError`
        """
        if not self.has_phone_number():
            raise SteamAuthenticatorError("Account doesn't have a verified phone number")

        resp = self._send_request('AddAuthenticator', {
            'steamid': self.backend.steam_id,
            'authenticator_time': int(time()),
            'authenticator_type': int(ETwoFactorTokenType.ValveMobileApp),
            'device_identifier': generate_device_id(self.backend.steam_id),
            'sms_phone_id': '1',
        })

        if resp['status'] != EResult.OK:
            raise SteamAuthenticatorError("Failed to add authenticator. Error: %s" % repr(EResult(resp['status'])))

        self.secrets = resp
        self.steam_time_offset = int(resp['server_time']) - time()

    def finalize(self, activation_code):
        """Finalize authenticator with received SMS code

        :param activation_code: SMS code
        :type activation_code: str
        :raises: :class:`SteamAuthenticatorError`
        """
        resp = self._send_request('FinalizeAddAuthenticator', {
            'steamid': self.backend.steam_id,
            'authenticator_time': int(time()),
            'authenticator_code': self.get_code(),
            'activation_code': activation_code,
        })

        if resp['status'] != EResult.TwoFactorActivationCodeMismatch and resp.get('want_more', False) and self._finalize_attempts:
            self.steam_time_offset += 30
            self._finalize_attempts -= 1
            self.finalize(activation_code)
            return
        elif not resp['success']:
            self._finalize_attempts = 5
            raise SteamAuthenticatorError("Failed to finalize authenticator. Error: %s" % repr(EResult(resp['status'])))

        self.steam_time_offset = int(resp['server_time']) - time()

    def remove(self, revocation_code=None):
        """Remove authenticator

        :param revocation_code: revocation code for account (e.g. R12345)
        :type  revocation_code: str

        .. note::
            After removing authenticator Steam Guard will be set to email codes

        .. warning::
            Doesn't work via :class:`.SteamClient`. Disabled by Valve

        :raises: :class:`SteamAuthenticatorError`
        """
        if not self.secrets:
            raise SteamAuthenticatorError("No authenticator secrets available?")
        if not isinstance(self.backend, MobileWebAuth):
            raise SteamAuthenticatorError("Only available via MobileWebAuth")

        resp = self._send_request('RemoveAuthenticator', {
            'steamid': self.backend.steam_id,
            'revocation_code': revocation_code if revocation_code else self.revocation_code,
            'steamguard_scheme': 1,
        })

        if not resp['success']:
            raise SteamAuthenticatorError("Failed to remove authenticator. (attempts remaining: %s)" % (
                resp['revocation_attempts_remaining'],
                ))

        self.secrets.clear()

    def status(self):
        """Fetch authenticator status for the account

        :raises: :class:`SteamAuthenticatorError`
        :return: dict with status parameters
        :rtype: dict
        """
        return self._send_request('QueryStatus', {'steamid': self.backend.steam_id})

    def create_emergency_codes(self, code=None):
        """Generate emergency codes

        :param code: SMS code
        :type code: str
        :raises: :class:`SteamAuthenticatorError`
        :return: list of codes
        :rtype: list

        .. note::
            A confirmation code is required to generate emergency codes and this method needs
            to be called twice as shown below.

        .. code:: python

            sa.create_emergency_codes()              # request a SMS code
            sa.create_emergency_codes(code='12345')  # creates emergency codes
        """
        if code:
            return self._send_request('createemergencycodes', {'code': code}).get('codes', [])
        else:
            self._send_request('createemergencycodes', {})
            return None

    def destroy_emergency_codes(self):
        """Destroy all emergency codes

        :raises: :class:`SteamAuthenticatorError`
        """
        self._send_request('DestroyEmergencyCodes', {'steamid': self.backend.steam_id})

    def _get_web_session(self):
        """
        :return: authenticated web session
        :rtype: :class:`requests.Session`
        :raises: :class:`RuntimeError` when session is unavailable
        """
        if isinstance(self.backend, MobileWebAuth):
            return self.backend.session
        else:
            if self.backend.logged_on:
                sess = self.backend.get_web_session()

                if sess is None:
                    raise RuntimeError("Failed to get a web session. Try again in a few minutes")
                else:
                    return sess
            else:
                raise RuntimeError("SteamClient instance is not connected")

    def add_phone_number(self, phone_number):
        """Add phone number to account

        Steps:

        1. Call :meth:`add_phone_number()` then check ``email_confirmation`` key in the response

           i. On ``True``, user needs to click link in email, then step 2
           ii. On ``False``, SMS code is sent,  go to step 3

        2. Confirm email via :meth:`confirm_email()`, SMS code is sent
        3. Finalize phone number with SMS code :meth:`confirm_phone_number(sms_code)`

        :param phone_number: phone number with country code
        :type  phone_number: :class:`str`
        :return: see example below
        :rtype: :class:`dict`

        .. code:: python

            {'success': True,
             'email_confirmation': True,
             'error_text': '',
             'fatal': False}
        """
        sess = self._get_web_session()

        try:
            resp = sess.post('https://steamcommunity.com/steamguard/phoneajax',
                             data={
                                 'op': 'add_phone_number',
                                 'arg': phone_number,
                                 'checkfortos': 0,
                                 'skipvoip': 0,
                                 'sessionid': sess.cookies.get('sessionid', domain='steamcommunity.com'),
                                 },
                             timeout=15).json()
        except:
            return {'success': False}

        return resp

    def confirm_email(self):
        """Confirm email confirmation. See :meth:`add_phone_number()`

        .. note::
            If ``email_confirmation`` is ``True``, then user hasn't clicked the link yet.

        :return: see example below
        :rtype: :class:`dict`

        .. code:: python

            {'success': True,
             'email_confirmation': True,
             'error_text': '',
             'fatal': False}
        """
        sess = self._get_web_session()

        try:
            resp = sess.post('https://steamcommunity.com/steamguard/phoneajax',
                             data={
                                 'op': 'email_confirmation',
                                 'arg': '',
                                 'checkfortos': 1,
                                 'skipvoip': 1,
                                 'sessionid': sess.cookies.get('sessionid', domain='steamcommunity.com'),
                                 },
                             timeout=15).json()
        except:
            return {'fatal': True, 'success': False}

        return resp

    def confirm_phone_number(self, sms_code):
        """Confirm phone number with the recieved SMS code. See :meth:`add_phone_number()`

        :param sms_code: sms code
        :type  sms_code: :class:`str`
        :return: see example below
        :rtype: :class:`dict`

        .. code:: python

            {'success': True,
             'error_text': '',
             'fatal': False}
        """
        sess = self._get_web_session()

        try:
            resp = sess.post('https://steamcommunity.com/steamguard/phoneajax',
                             data={
                                 'op': 'check_sms_code',
                                 'arg': sms_code,
                                 'checkfortos': 1,
                                 'skipvoip': 1,
                                 'sessionid': sess.cookies.get('sessionid', domain='steamcommunity.com'),
                                 },
                             timeout=15).json()
        except:
            return {'success': False}

        return resp

    def has_phone_number(self):
        """Check whether the account has a verified phone number

        :return: see example below
        :rtype: :class:`dict`

        .. code:: python

            {'success': True,
             'has_phone': True,
             'error_text': '',
             'fatal': False}
        """
        sess = self._get_web_session()

        try:
            resp = sess.post('https://steamcommunity.com/steamguard/phoneajax',
                             data={
                                 'op': 'has_phone',
                                 'arg': '0',
                                 'checkfortos': 0,
                                 'skipvoip': 1,
                                 'sessionid': sess.cookies.get('sessionid', domain='steamcommunity.com'),
                                 },
                             timeout=15).json()
        except:
            return {'success': False}

        return resp

    def validate_phone_number(self, phone_number):
        """Test whether phone number is valid for Steam

        :param phone_number: phone number with country code
        :type  phone_number: :class:`str`
        :return: see example below
        :rtype: :class:`dict`

        .. code:: python

            {'is_fixed': False,
             'is_valid': False,
             'is_voip': True,
             'number': '+1 123-555-1111',
             'success': True}
        """
        sess = self._get_web_session()

        try:
            resp = sess.post('https://store.steampowered.com/phone/validate',
                             data={
                                'phoneNumber': phone_number,
                                'sessionID': sess.cookies.get('sessionid', domain='store.steampowered.com'),
                                },
                             allow_redirects=False,
                             timeout=15).json()
        except:
            resp = {'success': False}

        return resp


class SteamAuthenticatorError(Exception):
    pass


def generate_twofactor_code(shared_secret):
    """Generate Steam 2FA code for login with current time

    :param shared_secret: authenticator shared shared_secret
    :type shared_secret: bytes
    :return: steam two factor code
    :rtype: str
    """
    return generate_twofactor_code_for_time(shared_secret, time() + (get_time_offset() or 0))

def generate_twofactor_code_for_time(shared_secret, timestamp):
    """Generate Steam 2FA code for timestamp

    :param shared_secret: authenticator shared secret
    :type shared_secret: bytes
    :param timestamp: timestamp to use, if left out uses current time
    :type timestamp: int
    :return: steam two factor code
    :rtype: str
    """
    hmac = hmac_sha1(bytes(shared_secret),
                     struct.pack('>Q', int(timestamp)//30))  # this will NOT stop working in 2038

    start = ord(hmac[19:20]) & 0xF
    codeint = struct.unpack('>I', hmac[start:start+4])[0] & 0x7fffffff

    charset = '23456789BCDFGHJKMNPQRTVWXY'
    code = ''

    for _ in range(5):
        codeint, i = divmod(codeint, len(charset))
        code += charset[i]

    return code

def generate_confirmation_key(identity_secret, tag, timestamp):
    """Generate confirmation key for trades. Can only be used once.

    :param identity_secret: authenticator identity secret
    :type identity_secret: bytes
    :param tag: tag identifies what the request, see list below
    :type tag: str
    :param timestamp: timestamp to use for generating key
    :type timestamp: int
    :return: confirmation key
    :rtype: bytes

    Tag choices:

        * ``conf`` to load the confirmations page
        * ``details`` to load details about a trade
        * ``allow`` to confirm a trade
        * ``cancel`` to cancel a trade

    """
    data = struct.pack('>Q', int(timestamp)) + tag.encode('ascii') # this will NOT stop working in 2038
    return hmac_sha1(bytes(identity_secret), data)

def get_time_offset():
    """Get time offset from steam server time via WebAPI

    :return: time offset (``None`` when Steam WebAPI fails to respond)
    :rtype: :class:`int`, :class:`None`
    """
    try:
        resp = webapi.post('ITwoFactorService', 'QueryTime', 1, params={'http_timeout': 10})
    except:
        return None

    ts = int(time())
    return int(resp.get('response', {}).get('server_time', ts)) - ts

def generate_device_id(steamid):
    """Generate Android device id

    :param steamid: Steam ID
    :type steamid: :class:`.SteamID`, :class:`int`
    :return: android device id
    :rtype: str
    """
    h = hexlify(sha1_hash(str(steamid).encode('ascii'))).decode('ascii')
    return "android:%s-%s-%s-%s-%s" % (h[:8], h[8:12], h[12:16], h[16:20], h[20:32])

def extract_secrets_from_android_rooted(adb_path='adb'):
    """Extract Steam Authenticator secrets from a rooted Android device

    Prerequisite for this to work:

        - rooted android device
        - `adb binary <https://developer.android.com/studio/command-line/adb.html>`_
        - device in debug mode, connected and paired

    .. note::
        If you know how to make this work, without requiring the device to be rooted,
        please open a issue on github. Thanks

    :param adb_path: path to adb binary
    :type adb_path: str
    :raises: When there is any problem
    :return: all secrets from the device, steamid as key
    :rtype: dict
    """
    data = subprocess.check_output([
        adb_path, 'shell', 'su', '-c',
        "'cat /data/data/com.valvesoftware.android.steam.community/files/Steamguard*'"
        ])

    # When adb daemon is not running, `adb` will print a couple of lines before our data.
    # The data doesn't have new lines and its always on the last line.
    data = data.decode('utf-8').split('\n')[-1]

    if data[0] != "{":
        raise RuntimeError("Got invalid data: %s" % repr(data))

    return {int(x['steamid']): x
            for x in map(json.loads, data.replace("}{", '}|||||{').split('|||||'))}