3 changed files with 0 additions and 519 deletions
@ -1,98 +0,0 @@ |
|||||
""" |
|
||||
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 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 |
|
||||
|
|
||||
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` |
|
||||
""" |
|
||||
|
|
||||
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 |
|
@ -1,220 +0,0 @@ |
|||||
# -*- 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 |
|
@ -1,201 +0,0 @@ |
|||||
# -*- 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 |
|
||||
import requests |
|
||||
|
|
||||
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicNumbers |
|
||||
from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15 |
|
||||
from steam.core.crypto import backend |
|
||||
|
|
||||
from steam.util.web import make_requests_session |
|
||||
from steam import SteamID |
|
||||
|
|
||||
if sys.version_info < (3,): |
|
||||
intBase = long |
|
||||
else: |
|
||||
intBase = int |
|
||||
|
|
||||
|
|
||||
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) |
|
||||
captcha_gid = -1 |
|
||||
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() |
|
||||
|
|
||||
@property |
|
||||
def captcha_url(self): |
|
||||
if self.captcha_gid == -1: |
|
||||
return None |
|
||||
else: |
|
||||
return "https://store.steampowered.com/login/rendercaptcha/?gid=%s" % self.captcha_gid |
|
||||
|
|
||||
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() |
|
||||
except requests.exceptions.RequestException as e: |
|
||||
raise HTTPError(str(e)) |
|
||||
|
|
||||
return resp |
|
||||
|
|
||||
def _load_key(self): |
|
||||
if not self.key: |
|
||||
resp = self.get_rsa_key(self.username) |
|
||||
|
|
||||
nums = RSAPublicNumbers(intBase(resp['publickey_exp'], 16), |
|
||||
intBase(resp['publickey_mod'], 16), |
|
||||
) |
|
||||
|
|
||||
self.key = backend.load_rsa_public_numbers(nums) |
|
||||
self.timestamp = resp['timestamp'] |
|
||||
|
|
||||
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) |
|
||||
self.session.cookies.set('mobileClient', 'android', domain=domain, secure=False) |
|
||||
self.session.cookies.set('Steam_Language', 'english', domain=domain, secure=False) |
|
||||
self.session.cookies.set('dob', '', domain=domain, secure=False) |
|
||||
|
|
||||
self._load_key() |
|
||||
|
|
||||
data = { |
|
||||
'username': self.username, |
|
||||
"password": b64encode(self.key.encrypt(self.password.encode('ascii'), PKCS1v15())), |
|
||||
"emailauth": email_code, |
|
||||
"emailsteamid": str(self.steamid) if email_code else '', |
|
||||
"twofactorcode": twofactor_code, |
|
||||
"captchagid": self.captcha_gid, |
|
||||
"captcha_text": captcha, |
|
||||
"loginfriendlyname": "python-steam webauth", |
|
||||
"rsatimestamp": self.timestamp, |
|
||||
"remember_login": 'true', |
|
||||
"donotcache": int(time() * 100000), |
|
||||
} |
|
||||
data['oauth_client_id'] = 'DE45CD61' |
|
||||
data['oauth_scope'] = 'read_profile write_profile read_client write_client' |
|
||||
data['loginfriendlyname'] = '#login_emailauth_friendlyname_mobile' |
|
||||
|
|
||||
try: |
|
||||
resp = self.session.post('https://steamcommunity.com/mobilelogin/dologin/', data=data, timeout=15).json() |
|
||||
except requests.exceptions.RequestException as e: |
|
||||
raise HTTPError(str(e)) |
|
||||
|
|
||||
self.captcha_gid = -1 |
|
||||
|
|
||||
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, resp['oauth']['wgtoken_secure']), |
|
||||
domain=domain, secure=True) |
|
||||
|
|
||||
return resp |
|
||||
else: |
|
||||
if resp.get('captcha_needed', False): |
|
||||
self.captcha_gid = resp['captcha_gid'] |
|
||||
|
|
||||
raise CaptchaRequired(resp['message']) |
|
||||
elif resp.get('emailauth_needed', False): |
|
||||
self.steamid = SteamID(resp['emailsteamid']) |
|
||||
raise EmailCodeRequired(resp['message']) |
|
||||
elif resp.get('requires_twofactor', False): |
|
||||
raise TwoFactorCodeRequired(resp['message']) |
|
||||
else: |
|
||||
raise LoginIncorrect(resp['message']) |
|
||||
|
|
||||
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 |
|
Loading…
Reference in new issue