From 253dcdd4cbc11f0714b9c038a0db47debae62f34 Mon Sep 17 00:00:00 2001 From: Philipp Joos Date: Fri, 3 Jun 2016 14:01:09 +0200 Subject: [PATCH] Cleanup --- requirements.txt | 1 + steam/authenticator.py | 174 +++++++++++++++++--------------- steam/guard.py | 4 +- steam/mobile.py | 220 +++++++++++++++++++++++++++++++++++++++++ steam/mobileauth.py | 116 +++++++++++++--------- steam/webauth.py | 23 +++-- 6 files changed, 403 insertions(+), 135 deletions(-) create mode 100644 steam/mobile.py diff --git a/requirements.txt b/requirements.txt index 638817c..9c3bc1e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ coverage==4.0.3 +setuptools>=11.3 gevent-eventemitter==1.4 enum34==1.1.2; python_version < '3.4' gevent==1.1.0 diff --git a/steam/authenticator.py b/steam/authenticator.py index 13f2e86..ee92b1e 100644 --- a/steam/authenticator.py +++ b/steam/authenticator.py @@ -1,86 +1,98 @@ +""" +This module provides methods for managing the mobile authenticator + +.. warning:: + Save your credentials! + +Example usage: + +.. code:: python + + import steam.authenticator + + ma = steam.authenticator.MobileAuthenticator('username', 'password') + credentials = ma.add_authenticator() + + sms_code = raw_input('SMS Code: ') + + ma.finalize_authenticator(sms_code) + + print credentials + +""" import json from guard import * -from mobileauth import MobileAuth - -class MobileAuthenticator: - def __init__(self, username, password, authenticatorCredentials=False): - self.username = username - self.password = password - self.ready = None - self.mobile = MobileAuth(username, password) - self.credentials = authenticatorCredentials or { } - - def login(self): - if self.ready != None: - return False - - if 'secret' in self.credentials.keys(): - code = generate_twofactor_code(self.credentials.get('secret')) - self.mobile.login(twofactor_code=code) - return True - else: - try: - self.mobile.login() - except EmailCodeRequired: - raise AuthenticatorAlreadyActive('Two factor authentication already active') - except TwoFactorCodeRequired - raise AuthenticatorAlreadyActive('Two factor authentication already active') - else: - self.ready = False - return True - - def addAuthenticator(self): - if self.ready != False: - return None - - - data = { - 'steamid': self.mobile.steamid, - 'sms_phone_id': 1, - 'access_token': self.mobile.oauth['oauth_token'], - 'authenticator_time': get_time_offset(), - 'authenticator_type': 1, - 'device_identifier': generate_device_id(self.mobile.steamid) - } - - response = self.mobile.request('https://api.steampowered.com/ITwoFactorService/AddAuthenticator/v1/', data) - if response.status_code == 200: - responseData = json.loads(response.text) - self.credentials = responseData['response'] - self.credentials['secret'] = self.credentials['uri'].split('?secret=')[1].split('&issuer')[0] - return True - else: - return [False, responseData] - - def finalizeAuthenticator(self, smsCode=None, tries=1): - if not smsCode or self.ready != False: - return None - - timestamp = get_time_offset() - - data = { - 'steamid': self.mobile.steamid, - 'access_token': self.mobile.oauth['oauth_token'], - 'authenticator_time': timestamp, - 'authenticator_code': generate_twofactor_code_for_time(self.credentials['secret'], timestamp), - 'activation_code': smsCode - } - response = self.mobile.request('https://api.steampowered.com/ITwoFactorService/FinalizeAddAuthenticator/v1/', data) - if response.status_code == 200: - responseData = json.loads(response.text) - if responseData['success']: - return True - else: - if responseData['want_more'] and tries < 30: - return self.finalizeAuthenticator(smsCode, tries) +from mobile import SteamMobile + + +class MobileAuthenticator(object): + mobile = None + + def __init__(self, username, password, login_mode='normal', email_code='', twofactor_code=''): + self.mobile = SteamMobile(username, password, login_mode, email_code, twofactor_code) + + def add_authenticator(self): + """Start process of linking a mobile authenticator to the logged in steam account + + :return: account credentials or False + :rtype: :class:`tuple`, :class:`bool` + """ + + data = { + 'steamid': self.mobile.steamid, + 'sms_phone_id': 1, + 'access_token': self.mobile.oauth['oauth_token'], + 'authenticator_time': get_time_offset(), + 'authenticator_type': 1, + 'device_identifier': generate_device_id(self.mobile.steamid) + } + + [status, body] = self.mobile.api_request('ITwoFactorService', 'AddAuthenticator', data, + return_including_status_code=True) + if status == 200: + responseData = json.loads(body) + self.credentials = responseData['response'] + self.credentials['secret'] = self.credentials['uri'].split('?secret=')[1].split('&issuer')[0] + return responseData else: - return False - else: - return False - -class MobileAuthenticatorException(Exception): - pass + return False + + def finalize_authenticator(self, sms_code=None, tries=1): + """Start process of linking a mobile authenticator to the logged in steam account + + :param sms_code: text reponse recieved by sms + :type sms_code: :class:`str` + :return: :class:`None` it no sms code is supplied, `True` or `False` + :rtype: :class:`None`, :class:`bool` + """ -class AuthenticatorAlreadyActive(MobileAuthenticatorException) - pass + if not sms_code: + return None + + timestamp = get_time_offset() + + data = { + 'steamid': self.mobile.steamid, + 'access_token': self.mobile.oauth['oauth_token'], + 'authenticator_time': timestamp, + 'authenticator_code': generate_twofactor_code_for_time(self.credentials['shared_secret'], timestamp), + 'activation_code': sms_code + } + [status, body] = self.mobile.api_request('ITwoFactorService', 'FinalizeAddAuthenticator', data, + return_including_status_code=True) + if status == 200: + responseData = json.loads(body)['response'] + if responseData['success']: + return True + else: + if responseData['want_more'] and tries < 30: + return self.finalizeAuthenticator(sms_code, tries) + else: + return False + else: + return False + + +class MobileAuthenticatorException(Exception): + pass \ No newline at end of file diff --git a/steam/guard.py b/steam/guard.py index 7d184f2..df3c185 100644 --- a/steam/guard.py +++ b/steam/guard.py @@ -1,9 +1,11 @@ import struct +import hashlib from binascii import hexlify from time import time from steam import webapi from steam.core.crypto import hmac_sha1, sha1_hash + def generate_twofactor_code(shared_secret): """Generate Steam 2FA code for login with current time @@ -84,5 +86,5 @@ def generate_device_id(steamid): :return: android device id :rtype: str """ - h = hexlify(sha1(str(steamid).encode('ascii'))).decode('ascii') + h = hexlify(str(hashlib.sha1(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]) diff --git a/steam/mobile.py b/steam/mobile.py new file mode 100644 index 0000000..8a1b7af --- /dev/null +++ b/steam/mobile.py @@ -0,0 +1,220 @@ +# -*- coding: utf-8 -*- +""" +This module provides utility functions to access steam mobile pages +""" + +import json +import requests + +import mobileauth + +API_ENDPOINTS = { + "IMobileAuthService": { + "methods": { + "GetWGToken": { + "version": 1 + } + } + }, + "ISteamWebUserPresenceOAuth": { + "methods": { + "Logon": { + "version": 1 + }, + "Logoff": { + "version": 1 + }, + "Message": { + "version": 1 + }, + "DeviceInfo": { + "version": 1 + }, + "Poll": { + "version": 1 + } + } + }, + "ISteamUserOAuth": { + "methods": { + "GetUserSummaries": { + "version": 1 + }, + "GetGroupSummaries": { + "version": 1 + }, + "GetGroupList": { + "version": 1 + }, + "GetFriendList": { + "version": 1 + }, + "Search": { + "version": 1 + } + } + }, + + "ISteamGameOAuth": { + "methods": { + "GetAppInfo": { + "version": 1 + } + } + }, + "IMobileNotificationService": { + "methods": { + "SwitchSessionToPush": { + "version": 1 + } + } + }, + "IFriendMessagesService": { + "methods": { + "GetActiveMessageSessions": { + "version": 1 + }, + "GetRecentMessages": { + "version": 1 + }, + "MarkOfflineMessagesRead": { + "version": 1 + } + } + }, + "ITwoFactorService": { + "methods": { + "AddAuthenticator": { + "version": 1 + }, + "RecoverAuthenticatorCommit": { + "version": 1 + }, + "RecoverAuthenticatorContinue": { + "version": 1 + }, + "RemoveAuthenticator": { + "version": 1 + }, + "RemoveAuthenticatorViaChallengeStart": { + "version": 1 + }, + "RemoveAuthenticatorViaChallengeContinue": { + "version": 1 + }, + "FinalizeAddAuthenticator": { + "version": 1 + }, + "QueryStatus": { + "version": 1 + }, + "QueryTime": { + "version": 1 + }, + "QuerySecrets": { + "version": 1 + }, + "SendEmail": { + "version": 1 + }, + "ValidateToken": { + "version": 1 + }, + "CreateEmergencyCodes": { + "version": 1 + }, + "DestroyEmergencyCodes": { + "version": 1 + } + } + } +} + +API_BASE_URL = 'https://api.steampowered.com' + + +class SteamMobile(object): + session = None + oauth = None + steamid = None + + def __init__(self, username, password, login_mode='normal', email_code='', twofactor_code=''): + mobile_auth = mobileauth.MobileAuth(username, password) + + try: + if login_mode == 'normal': + mobile_auth.login() + elif login_mode == 'email': + mobile_auth.login(email_code=email_code) + elif login_mode == 'twofa': + mobile_auth.login(twofactor_code=twofactor_code) + + except mobileauth.CaptchaRequired: + raise CaptchaNotSupported("Captcha's are currently not supported. Please wait a few minutes before you try to login again.") + except mobileauth.EmailCodeRequired: + mobile_auth.login(email_code=email_code) + except mobileauth.TwoFactorCodeRequired: + mobile_auth.login(twofactor_code=twofactor_code) + + + self.session = mobile_auth.session + self.steamid = mobile_auth.steamid + self.oauth = mobile_auth.oauth + + def refresh_session(self, oauth_token=None): + oauth_token = oauth_token or self.oauth['oauth_token'] + response = self.api_request('IMobileAuthService', 'GetWGToken', {'access_token': oauth_token}) + try: + data = json.loads(response) + except Exception, e: + raise SessionRefreshFailed(str(e)) + else: + self.oauth['wgtoken'] = data['response']['token'] + self.oauth['wgtoken_secure'] = data['response']['token_secure'] + for domain in ['store.steampowered.com', 'help.steampowered.com', 'steamcommunity.com']: + self.session.cookies.set('steamLogin', '%s||%s' % (self.steamid, self.oauth['wgtoken']), domain=domain, + secure=False) + self.session.cookies.set('steamLoginSecure', '%s||%s' % (self.steamid, self.oauth['wgtoken_secure']), + domain=domain, secure=True) + + def _request(self, uri, data={}, return_including_status_code=False): + headers = { + 'X-Requested-With': 'com.valvesoftware.android.steam.community', + 'User-agent': 'Mozilla/5.0 (Linux; U; Android 4.1.1; en-us; Google Nexus 4 - 4.1.1 - API 16 - 768x1280 Build/JRO03S)\ + AppleWebKit/534.30 (KHTML, like Gecko) Version/4.0 Mobile Safari/534.30' + } + + try: + response = self.session.post(uri, data=data, headers=headers) + except requests.exceptions.RequestException as e: + raise mobileauth.HTTPError(str(e)) + else: + if return_including_status_code: + return [response.status_code, response.text] + else: + return response.text + + def api_request(self, interface, method, data={}, return_including_status_code=False): + if interface in API_ENDPOINTS.keys() and method in API_ENDPOINTS.get(interface).get('methods').keys(): + uri = '%s/%s/%s/v%s/' % ( + API_BASE_URL, interface, method, API_ENDPOINTS.get(interface).get('methods').get(method).get('version')) + response = self._request(uri, data, return_including_status_code) + return response + else: + raise APIEndpointNotFound('Endpoint %s.%s not found' % (interface, method)) + + +class SteamMobileException(Exception): + pass + + +class CaptchaNotSupported(SteamMobileException): + pass + + +class SessionRefreshFailed(SteamMobileException): + pass + + +class APIEndpointNotFound(SteamMobileException): + pass \ No newline at end of file diff --git a/steam/mobileauth.py b/steam/mobileauth.py index 8ac2b0d..3004496 100644 --- a/steam/mobileauth.py +++ b/steam/mobileauth.py @@ -1,4 +1,54 @@ # -*- coding: utf-8 -*- +""" +This module simplifies the process of obtaining an authenticated session for mobile steam websites. +After authentication is complete, a :class:`requests.Session` is created containing the auth and mobile specific cookies. +The session can be used to access ``steamcommunity.com``, ``store.steampowered.com``, and ``help.steampowered.com``. +The main purpose of a mobile session, is to access the mobile trade confirmations page and to easily access the +ITwoFactorService api, but can also used to access all 'normal' steam pages. + + +.. warning:: + A web session may expire randomly, or when you login from different IP address. + Some pages will return status code `401` when that happens. + Keep in mind if you are trying to write robust code. + +Example usage: + +.. code:: python + + import steam.mobileauth as ma + + user = ma.WebAuth('username', 'password') + + try: + user.login() + except wa.CaptchaRequired: + print user.captcha_url + # ask a human to solve captcha + user.login(captcha='ABC123') + except wa.EmailCodeRequired: + user.login(email_code='ZXC123') + except wa.TwoFactorCodeRequired: + user.login(twofactor_code='ZXC123') + + user.session.get('https://store.steampowered.com/account/history/') + # OR + session = user.login() + session.get('https://store.steampowered.com/account/history') + +Alternatively, if Steam Guard is not enabled on the account: + +.. code:: python + + try: + session = wa.WebAuth('username', 'password').login() + except wa.HTTPError: + pass + +The :class:`MobileAuth` instance should be discarded once a session is obtained +as it is not reusable. +""" +import json from time import time import sys from base64 import b64encode @@ -20,10 +70,11 @@ else: class MobileAuth(object): key = None complete = False #: whether authentication has been completed successfully - session = None #: :class:`requests.Session` (with auth cookies after auth is complete) + session = None #: :class:`requests.Session` (with auth cookies after auth is complete) captcha_gid = -1 - steamid = None #: :class:`steam.steamid.SteamID` (after auth is complete) + steamid = None #: :class:`steam.steamid.SteamID` (after auth is complete) oauth = {} + def __init__(self, username, password): self.__dict__.update(locals()) self.session = make_requests_session() @@ -38,12 +89,12 @@ class MobileAuth(object): def get_rsa_key(self, username): try: resp = self.session.post('https://steamcommunity.com/mobilelogin/getrsakey/', - timeout=15, - data={ - 'username': username, - 'donotchache': int(time() * 1000), - }, - ).json() + timeout=15, + data={ + 'username': username, + 'donotchache': int(time() * 1000), + }, + ).json() except requests.exceptions.RequestException as e: raise HTTPError(str(e)) @@ -59,40 +110,11 @@ class MobileAuth(object): self.key = backend.load_rsa_public_numbers(nums) self.timestamp = resp['timestamp'] - - def request(self, uri, data): - if not self.complete: - return None - - headers = { - 'X-Requested-With': 'com.valvesoftware.android.steam.community', - 'User-agent': 'Mozilla/5.0 (Linux; U; Android 4.1.1; en-us; Google Nexus 4 - 4.1.1 - API 16 - 768x1280 Build/JRO03S) AppleWebKit/534.30 (KHTML, like Gecko) Version/4.0 Mobile Safari/534.30' - } - - try: - response = self.session.post(uri, data=data, headers=headers) - except requests.exceptions.RequestException as e: - raise HTTPError(str(e)) - else: - return response - - def refreshSession(self, oauth_token=None): - oauth_token = oauth_token or self.oauth['oauth_token'] - response = self.request('https://api.steampowered.com/IMobileAuthService/GetWGToken/v0001', {'access_token': oauth_token}) - try: - data = json.loads(response) - except Exception, e: - raise RefreshSessionFailed(str(e)) - else: - self.oauth['wgtoken'] = data['response']['token'] - self.oauth['wgtoken_secure'] = data['response']['token_secure'] - self.session.cookies.set('steamLogin', '%s||%s' % (self.steamid, sself.oauth['wgtoken']), domain=domain, secure=False) - self.session.cookies.set('steamLoginSecure', '%s||%s' % (self.steamid, self.oauth['wgtoken_secure']), domain=domain, secure=True) def login(self, captcha='', email_code='', twofactor_code='', language='english'): if self.complete: return self.session - + for domain in ['store.steampowered.com', 'help.steampowered.com', 'steamcommunity.com']: self.session.cookies.set('forceMobile', '1', domain=domain, secure=False) self.session.cookies.set('mobileClientVersion', '0 (2.1.3)', domain=domain, secure=False) @@ -103,7 +125,7 @@ class MobileAuth(object): self._load_key() data = { - 'username' : self.username, + 'username': self.username, "password": b64encode(self.key.encrypt(self.password.encode('ascii'), PKCS1v15())), "emailauth": email_code, "emailsteamid": str(self.steamid) if email_code else '', @@ -129,11 +151,14 @@ class MobileAuth(object): if resp['success'] and resp['login_complete']: self.complete = True self.password = None + resp['oauth'] = json.loads(resp['oauth']) self.steamid = SteamID(resp['oauth']['steamid']) self.oauth = resp['oauth'] for domain in ['store.steampowered.com', 'help.steampowered.com', 'steamcommunity.com']: - self.session.cookies.set('steamLogin', '%s||%s' % (self.steamid, resp['oauth']['wgtoken']), domain=domain, secure=False) - self.session.cookies.set('steamLoginSecure', '%s||%s' % (self.steamid, data['oauth']['wgtoken_secure']), domain=domain, secure=True) + self.session.cookies.set('steamLogin', '%s||%s' % (self.steamid, resp['oauth']['wgtoken']), + domain=domain, secure=False) + self.session.cookies.set('steamLoginSecure', '%s||%s' % (self.steamid, resp['oauth']['wgtoken_secure']), + domain=domain, secure=True) return resp else: @@ -151,23 +176,26 @@ class MobileAuth(object): return None + class MobileWebAuthException(Exception): pass + class HTTPError(MobileWebAuthException): pass + class LoginIncorrect(MobileWebAuthException): pass + class CaptchaRequired(MobileWebAuthException): pass + class EmailCodeRequired(MobileWebAuthException): pass -class TwoFactorCodeRequired(MobileWebAuthException): - pass -class RefreshSessionFailed(MobileWebAuthException): +class TwoFactorCodeRequired(MobileWebAuthException): pass diff --git a/steam/webauth.py b/steam/webauth.py index 4e7d57b..3cc9513 100644 --- a/steam/webauth.py +++ b/steam/webauth.py @@ -70,9 +70,9 @@ else: class WebAuth(object): key = None complete = False #: whether authentication has been completed successfully - session = None #: :class:`requests.Session` (with auth cookies after auth is complete) + session = None #: :class:`requests.Session` (with auth cookies after auth is complete) captcha_gid = -1 - steamid = None #: :class:`steam.steamid.SteamID` (after auth is complete) + steamid = None #: :class:`steam.steamid.SteamID` (after auth is complete) def __init__(self, username, password): self.__dict__.update(locals()) @@ -97,12 +97,12 @@ class WebAuth(object): """ try: resp = self.session.post('https://store.steampowered.com/login/getrsakey/', - timeout=15, - data={ - 'username': username, - 'donotchache': int(time() * 1000), - }, - ).json() + timeout=15, + data={ + 'username': username, + 'donotchache': int(time() * 1000), + }, + ).json() except requests.exceptions.RequestException as e: raise HTTPError(str(e)) @@ -144,7 +144,7 @@ class WebAuth(object): self._load_key() data = { - 'username' : self.username, + 'username': self.username, "password": b64encode(self.key.encrypt(self.password.encode('ascii'), PKCS1v15())), "emailauth": email_code, "emailsteamid": str(self.steamid) if email_code else '', @@ -199,17 +199,22 @@ class WebAuth(object): class WebAuthException(Exception): pass + class HTTPError(WebAuthException): pass + class LoginIncorrect(WebAuthException): pass + class CaptchaRequired(WebAuthException): pass + class EmailCodeRequired(WebAuthException): pass + class TwoFactorCodeRequired(WebAuthException): pass