-
Notifications
You must be signed in to change notification settings - Fork 151
/
confirmation.py
122 lines (102 loc) · 5.24 KB
/
confirmation.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
import enum
import json
import time
from typing import List
import requests
from bs4 import BeautifulSoup
from steampy import guard
from steampy.exceptions import ConfirmationExpected
from steampy.login import InvalidCredentials
class Confirmation:
def __init__(self, _id, data_confid, data_key):
self.id = _id.split('conf')[1]
self.data_confid = data_confid
self.data_key = data_key
class Tag(enum.Enum):
CONF = 'conf'
DETAILS = 'details'
ALLOW = 'allow'
CANCEL = 'cancel'
class ConfirmationExecutor:
CONF_URL = "https://steamcommunity.com/mobileconf"
def __init__(self, identity_secret: str, my_steam_id: str, session: requests.Session) -> None:
self._my_steam_id = my_steam_id
self._identity_secret = identity_secret
self._session = session
def send_trade_allow_request(self, trade_offer_id: str) -> dict:
confirmations = self._get_confirmations()
confirmation = self._select_trade_offer_confirmation(confirmations, trade_offer_id)
return self._send_confirmation(confirmation)
def confirm_sell_listing(self, asset_id: str) -> dict:
confirmations = self._get_confirmations()
confirmation = self._select_sell_listing_confirmation(confirmations, asset_id)
return self._send_confirmation(confirmation)
def _send_confirmation(self, confirmation: Confirmation) -> dict:
tag = Tag.ALLOW
params = self._create_confirmation_params(tag.value)
params['op'] = tag.value,
params['cid'] = confirmation.data_confid
params['ck'] = confirmation.data_key
headers = {'X-Requested-With': 'XMLHttpRequest'}
return self._session.get(self.CONF_URL + '/ajaxop', params=params, headers=headers).json()
def _get_confirmations(self) -> List[Confirmation]:
confirmations = []
confirmations_page = self._fetch_confirmations_page()
soup = BeautifulSoup(confirmations_page.text, 'html.parser')
if soup.select('#mobileconf_empty'):
return confirmations
for confirmation_div in soup.select('#mobileconf_list .mobileconf_list_entry'):
_id = confirmation_div['id']
data_confid = confirmation_div['data-confid']
data_key = confirmation_div['data-key']
confirmations.append(Confirmation(_id, data_confid, data_key))
return confirmations
def _fetch_confirmations_page(self) -> requests.Response:
tag = Tag.CONF.value
params = self._create_confirmation_params(tag)
headers = {'X-Requested-With': 'com.valvesoftware.android.steam.community'}
response = self._session.get(self.CONF_URL + '/conf', params=params, headers=headers)
if 'Steam Guard Mobile Authenticator is providing incorrect Steam Guard codes.' in response.text:
raise InvalidCredentials('Invalid Steam Guard file')
return response
def _fetch_confirmation_details_page(self, confirmation: Confirmation) -> str:
tag = 'details' + confirmation.id
params = self._create_confirmation_params(tag)
response = self._session.get(self.CONF_URL + '/details/' + confirmation.id, params=params)
return response.json()['html']
def _create_confirmation_params(self, tag_string: str) -> dict:
timestamp = int(time.time())
confirmation_key = guard.generate_confirmation_key(self._identity_secret, tag_string, timestamp)
android_id = guard.generate_device_id(self._my_steam_id)
return {'p': android_id,
'a': self._my_steam_id,
'k': confirmation_key,
't': timestamp,
'm': 'android',
'tag': tag_string}
def _select_trade_offer_confirmation(self, confirmations: List[Confirmation], trade_offer_id: str) -> Confirmation:
for confirmation in confirmations:
confirmation_details_page = self._fetch_confirmation_details_page(confirmation)
confirmation_id = self._get_confirmation_trade_offer_id(confirmation_details_page)
if confirmation_id == trade_offer_id:
return confirmation
raise ConfirmationExpected
def _select_sell_listing_confirmation(self, confirmations: List[Confirmation], asset_id: str) -> Confirmation:
for confirmation in confirmations:
confirmation_details_page = self._fetch_confirmation_details_page(confirmation)
confirmation_id = self._get_confirmation_sell_listing_id(confirmation_details_page)
if confirmation_id == asset_id:
return confirmation
raise ConfirmationExpected
@staticmethod
def _get_confirmation_sell_listing_id(confirmation_details_page: str) -> str:
soup = BeautifulSoup(confirmation_details_page, 'html.parser')
scr_raw = soup.select("script")[2].string.strip()
scr_raw = scr_raw[scr_raw.index("'confiteminfo', ") + 16:]
scr_raw = scr_raw[:scr_raw.index(", UserYou")].replace("\n", "")
return json.loads(scr_raw)["id"]
@staticmethod
def _get_confirmation_trade_offer_id(confirmation_details_page: str) -> str:
soup = BeautifulSoup(confirmation_details_page, 'html.parser')
full_offer_id = soup.select('.tradeoffer')[0]['id']
return full_offer_id.split('_')[1]