The new woob repository is here: https://gitlab.com/woob/woob. This gitlab will be removed soon.

Commit 78ab37de authored by Ilyas Semmaoui's avatar Ilyas Semmaoui Committed by Vincent A

[amazon] Prevent sms spam and remove restricted URL to not trigger sms sending

Previously we were sending 1 sms every minutes until the connection is approved
by the user. Moreover there was a restricted URL we were trying to access which
led us to trigger the in-app validation which sends an sms/e-mail/in-app
notification to approve the connection even if we were already logged in.
parent 1edd3628
......@@ -26,14 +26,15 @@
from woob.exceptions import (
BrowserIncorrectPassword, BrowserUnavailable, ImageCaptchaQuestion, BrowserQuestion,
WrongCaptchaResponse, NeedInteractiveFor2FA, BrowserPasswordExpired,
AppValidation, AppValidationExpired,
AppValidation, AppValidationExpired, AppValidationCancelled,
)
from woob.tools.value import Value
from woob.browser.browsers import ClientError
from .pages import (
LoginPage, SubscriptionsPage, DocumentsPage, DownloadDocumentPage, HomePage,
PanelPage, SecurityPage, LanguagePage, HistoryPage, PasswordExpired, ApprovalPage,
SecurityPage, LanguagePage, HistoryPage, PasswordExpired, ApprovalPage, PollingPage,
ResetPasswordPage,
)
......@@ -54,17 +55,22 @@ class AmazonBrowser(LoginBrowser, StatesMixin):
WRONG_CAPTCHA_RESPONSE = "Saisissez les caractères tels qu'ils apparaissent sur l'image."
login = URL(r'/ap/signin(.*)', LoginPage)
home = URL(r'/$', r'/\?language=\w+$', HomePage)
panel = URL('/gp/css/homepage.html/ref=nav_youraccount_ya', PanelPage)
subscriptions = URL(r'/ap/cnep(.*)', SubscriptionsPage)
history = URL(r'/gp/your-account/order-history\?ref_=ya_d_c_yo', HistoryPage)
home = URL(r'/$', r'/\?language=.+$', HomePage)
subscriptions = URL(r'/gp/profile', SubscriptionsPage)
history = URL(
r'/gp/your-account/order-history\?ref_=ya_d_c_yo',
r'/gp/css/order-history\?',
HistoryPage,
)
documents = URL(
r'/gp/your-account/order-history\?opt=ab&digitalOrders=1(.*)&orderFilter=year-(?P<year>.*)',
r'/gp/your-account/order-history',
DocumentsPage,
)
download_doc = URL(r'/gp/shared-cs/ajax/invoice/invoice.html', DownloadDocumentPage)
approval_page = URL(r'/ap/cvf/approval', ApprovalPage)
approval_page = URL(r'/ap/cvf/approval\?', ApprovalPage)
reset_password_page = URL(r'/ap/forgotpassword/reverification', ResetPasswordPage)
poll_page = URL(r'/ap/cvf/approval/poll', PollingPage)
security = URL(
r'/ap/dcq',
r'/ap/cvf/',
......@@ -76,7 +82,9 @@ class AmazonBrowser(LoginBrowser, StatesMixin):
__states__ = ('otp_form', 'otp_url', 'otp_style', 'otp_headers')
STATE_DURATION = 10
# According to the cookies we are fine for 1 year after the last sync.
# If we reset the state every 10 minutes we'll get a in-app validation after 10 minutes
STATE_DURATION = 60 * 24 * 365
otp_form = None
otp_url = None
......@@ -156,27 +164,39 @@ def handle_captcha(self, captcha):
raise ImageCaptchaQuestion(image)
def check_app_validation(self):
# client has 60 seconds to unlock this page
# the resend link will appear from 60 seconds is why there are 2 additional seconds, it's to have a margin
timeout = time.time() + 62.00
second_try = True
# 25' on website, we don't wait that much, but leave sufficient time for the user
timeout = time.time() + 600.00
app_validation_link = self.page.get_link_app_validation()
polling_request = self.page.get_polling_request()
approval_status = ''
while time.time() < timeout:
link = self.page.get_link_app_validation()
self.location(link)
if self.approval_page.is_here():
time.sleep(2)
else:
return
self.location(polling_request)
approval_status = self.page.get_approval_status()
if time.time() >= timeout and second_try:
# second try because 60 seconds is short, the second try is longger
second_try = False
timeout = time.time() + 70.00
self.page.resend_link()
if approval_status != 'TransactionPending':
break
# poll every 5 seconds on website
time.sleep(5)
else:
raise AppValidationExpired()
if approval_status in ['TransactionCompleted', 'TransactionResponded']:
self.location(app_validation_link)
if self.reset_password_page.is_here():
raise AppValidationCancelled()
if self.approval_page.is_here():
raise AssertionError('The validation was not effective for an unknown reason.')
elif approval_status == 'TransactionCompletionTimeout':
raise AppValidationExpired()
else:
raise AssertionError('Unknown transaction status: %s' % approval_status)
def do_login(self):
if self.config['pin_code'].get():
# Resolve pin_code
......@@ -279,18 +299,10 @@ def to_english(self, language):
@need_login
def iter_subscription(self):
self.location(self.panel.go().get_sub_link())
if self.home.is_here():
if self.page.get_login_link():
self.is_login()
self.location(self.page.get_panel_link())
elif not self.subscriptions.is_here():
self.is_login()
self.subscriptions.go()
# goes back to the subscription page as you may be redirected to the documents page
if not self.subscriptions.is_here():
self.location(self.panel.go().get_sub_link())
self.is_login()
yield self.page.get_item()
......
......@@ -23,28 +23,24 @@
from woob.browser.elements import ItemElement, ListElement, method
from woob.browser.filters.html import Link, Attr
from woob.browser.filters.standard import (
CleanText, CleanDecimal, Env, Regexp, Format,
Field, Currency, RegexpError, Date, Async, AsyncLoad,
CleanText, CleanDecimal, Env, Regexp, Format, RawText,
Field, Currency, Date, Async, AsyncLoad,
Coalesce,
)
from woob.capabilities.bill import DocumentTypes, Bill, Subscription
from woob.capabilities.base import NotAvailable
from woob.tools.json import json
from woob.tools.date import parse_french_date
class HomePage(HTMLPage):
def get_login_link(self):
return self.doc.xpath('//a[./span[contains(., "%s")]]/@href' % self.browser.L_SIGNIN)[0]
return Attr('//a[@data-nav-role="signin"]', 'href')(self.doc)
def get_panel_link(self):
return Link('//a[contains(@href, "homepage.html") and has-class(@nav-link)]')(self.doc)
class PanelPage(LoggedPage, HTMLPage):
def get_sub_link(self):
return CleanText('//a[@class="ya-card__whole-card-link" and contains(@href, "cnep")]/@href')(self.doc)
class SecurityPage(HTMLPage):
def get_otp_type(self):
if self.doc.xpath('//form[@id="auth-select-device-form"]'):
......@@ -108,18 +104,30 @@ def has_form_select_device(self):
class ApprovalPage(HTMLPage, LoggedPage):
def get_msg_app_validation(self):
msg = CleanText('//span[has-class("transaction-approval-word-break")]')
msg = CleanText('//div[has-class("a-spacing-large")]/span[has-class("transaction-approval-word-break")]')
sending_address = CleanText('//div[@class="a-row"][1]')
msg = Format('%s %s', msg, sending_address)
return msg(self.doc)
return Format('%s %s', msg, sending_address)(self.doc)
def get_link_app_validation(self):
return Link('//a[@id="resend-approval-link"]')(self.doc)
return Attr('//input[@name="openid.return_to"]', 'value')(self.doc)
def resend_link(self):
form = self.get_form(id='resend-approval-form')
form.submit()
def get_polling_request(self):
form = self.get_form(id="pollingForm")
return form.request
class PollingPage(HTMLPage):
def get_approval_status(self):
return Attr('//input[@name="transactionApprovalStatus"]', 'value', default=None)(self.doc)
class ResetPasswordPage(HTMLPage):
pass
class LanguagePage(HTMLPage):
pass
......@@ -165,14 +173,15 @@ class SubscriptionsPage(LoggedPage, HTMLPage):
class get_item(ItemElement):
klass = Subscription
def obj_subscriber(self):
try:
return Regexp(CleanText('//div[contains(@class, "a-fixed-right-grid-col")]'), self.page.browser.L_SUBSCRIBER)(self)
except RegexpError:
return self.page.browser.username
obj_id = 'amazon'
def obj_subscriber(self):
profile_data = json.loads(Regexp(
RawText('//script[contains(text(), "window.CustomerProfileRootProps")]'),
r'window.CustomerProfileRootProps = ({.+});',
)(self))
return profile_data.get('nameHeaderData', {}).get('name', NotAvailable)
def obj_label(self):
return self.page.browser.username
......@@ -219,7 +228,7 @@ def obj_date(self):
Date(CleanText('.//div[has-class("a-span2") and not(has-class("recipient"))]/div[2]'), parse_func=parse_french_date, dayfirst=True, default=NotAvailable),
)(self)
def obj_price(self):
def obj_total_price(self):
# Some orders, audiobooks for example, are paid using "audio credits", they have no price or currency
currency = Env('currency')(self)
return CleanDecimal(
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment