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

    sa = SteamAuthenticator(medium=medium)
    sa.add()    # SMS code will be send to account's phone number
    sa.secrets  # dict with authenticator secrets, make sure you save them

    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

    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.
"""
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.util import proto_to_dict


class SteamAuthenticator(object):
    """Add/Remove authenticator from an account. Generate 2FA and confirmation codes."""
    _finalize_attempts = 5
    medium =  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, medium=None):
        """
        :param secret: a dict of authenticator secrets
        :type secret: dict
        :param medium: logged on session for steam user
        :type mediumm: :class:`.MobileWebAuth`, :class:`.SteamClient`
        """
        self.secrets = secrets or {}
        self.medium = medium

    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):
        medium = self.medium

        if isinstance(medium, MobileWebAuth):
            if not medium.complete:
                raise SteamAuthenticatorError("MobileWebAuth instance not logged in")

            params['access_token'] = medium.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 medium.logged_on:
                raise SteamAuthenticatorError("SteamClient instance not logged in")

            resp = medium.unified_messages.send_and_wait("TwoFactor.%s#1" % action,
                                                         params, timeout=10)
            if resp is None:
                raise SteamAuthenticatorError("Failed to add authenticator. Request timeout")

            resp = proto_to_dict(resp)

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

        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.medium.steam_id,
            'authenticator_time': int(time()),
            'authenticator_type': int(ETwoFactorTokenType.ValveMobileApp),
            'device_identifier': generate_device_id(self.medium.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.medium.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):
        """Remove authenticator

        .. 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.medium, MobileWebAuth):
            raise SteamAuthenticatorError("Only available via MobileWebAuth")

        resp = self._send_request('RemoveAuthenticator', {
            'steamid': self.medium.steam_id,
            'revocation_code': 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, medium=None):
        """Fetch authenticator status for the account

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

    def create_emergency_codes(self):
        """Generate emergency codes

        :raises: :class:`SteamAuthenticatorError`
        :return: list of codes
        :rtype: list
        """
        return self._send_request('CreateEmergencyCodes', {}).get('code', [])

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

        :raises: :class:`SteamAuthenticatorError`
        """
        self._send_request('DestroyEmergencyCodes', {'steamid': self.medium.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.medium, MobileWebAuth):
            return self.medium.session
        else:
            if self.medium.logged_on:
                sess = self.medium.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

        Then confirm it via :meth:`confirm_phone_number()`

        :param phone_number: phone number with country code
        :type  phone_number: :class:`str`
        :return: success (returns ``False`` on request fail/timeout)
        :rtype: :class:`bool`
        """
        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 False

        return (resp or {}).get('success', False)

    def confirm_phone_number(self, sms_code):
        """Confirm phone number with the recieved SMS code

        :param sms_code: sms code
        :type  sms_code: :class:`str`
        :return: success (returns ``False`` on request fail/timeout)
        :rtype: :class:`bool`
        """
        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 False

        return (resp or {}).get('success', False)

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

        :return: result
        :rtype: :class:`bool` or :class:`None`

        .. note::
            Retruns `None` if the web requests fails for any reason
        """
        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:
            raise

        if resp['success'] == True:
            return resp['has_phone']

    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 output 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.get('https://store.steampowered.com//phone/validate',
                            params={'phoneNumber': phone_number},
                            timeout=15).json()
        except:
            resp = {}

        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 a 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*'"
        ])

    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('|||||'))}