Skip to content
Commits on Source (64)
......@@ -103,7 +103,7 @@ def has_form_select_device(self):
return bool(self.doc.xpath('//form[@id="auth-select-device-form"]'))
class ApprovalPage(HTMLPage, LoggedPage):
class ApprovalPage(HTMLPage):
def get_msg_app_validation(self):
msg = CleanText('//div[has-class("a-spacing-large")]/span[has-class("transaction-approval-word-break")]')
sending_address = CleanText('//div[@class="a-row"][1]')
......
......@@ -24,6 +24,9 @@
from dateutil.parser import parse as parse_date
from collections import OrderedDict
from woob.browser.selenium import (
SeleniumBrowser, SubSeleniumMixin, IsHereCondition, webdriver,
)
from woob.exceptions import (
BrowserIncorrectPassword, ActionNeeded, BrowserUnavailable,
AuthMethodNotImplemented, BrowserQuestion, ScrapingBlocked,
......@@ -39,14 +42,12 @@
JsonBalances2, CurrencyPage, LoginPage, NoCardPage,
NotFoundPage, HomeLoginPage,
ReadAuthChallengePage, UpdateAuthTokenPage,
SLoginPage,
)
from .fingerprint import FingerprintPage
__all__ = ['AmericanExpressBrowser']
class AmericanExpressBrowser(TwoFactorBrowser):
BASEURL = 'https://global.americanexpress.com'
TWOFA_BASEURL = r'https://functions.americanexpress.com'
......@@ -109,12 +110,10 @@ def __init__(self, *args, **kwargs):
}
def init_login(self):
self.home_login.go()
self.setup_browser_for_login_request()
transaction_id = self.make_transaction_id()
now = datetime.datetime.utcnow()
transaction_id = 'LOGIN-%s' % uuid.uuid4() # Randomly generated in js
self.register_transaction_id(transaction_id, now)
data = {
'request_type': 'login',
......@@ -206,9 +205,19 @@ def clear_init_cookies(self):
if device:
self.session.cookies.set_cookie(device)
def register_transaction_id(self, transaction_id, now):
def setup_browser_for_login_request(self):
self.home_login.go()
def make_transaction_id(self):
transaction_id = 'LOGIN-%s' % uuid.uuid4() # Randomly generated in js
self.register_transaction_id(transaction_id)
return transaction_id
def register_transaction_id(self, transaction_id):
self.fingerprint.go(transaction_id=transaction_id)
payload = self.page.make_payload_for_s2(transaction_id, now)
payload = self.page.make_payload_for_s2(transaction_id)
self.open('https://www.cdn-path.com/s2', method="POST",
params={
't': self.page.get_t(),
......@@ -473,3 +482,82 @@ def iter_coming(self, account):
yield tr
else:
return
class AmericanExpressSeleniumFingerprintBrowser(SeleniumBrowser):
BASEURL = 'https://global.americanexpress.com'
home_login = URL(r'/login\?inav=fr_utility_logout')
login = URL(r'https://www.americanexpress.com/en-us/account/login', SLoginPage)
HEADLESS = True # Always change to True for prod
WINDOW_SIZE = (1800, 1000)
DRIVER = webdriver.Chrome
def __init__(self, config, *args, **kwargs):
super(AmericanExpressSeleniumFingerprintBrowser, self).__init__(*args, **kwargs)
def do_login(self):
"""
We don't really support login via selenium. We only load the login to execute some
javascript and then extract cookies + some other values generated in javascript.
"""
self.home_login.go()
self.wait_until(IsHereCondition(self.login))
class AmericanExpressWithSeleniumBrowser(SubSeleniumMixin, AmericanExpressBrowser):
"""
Use a selenium browser to pass the fingerprinting instead of trying to solve it
manually.
Selenium is executed at the start of init_login in setup_browser_for_login_request.
From inside SubSeleniumMixin.do_login, the load_selenium_session method will be called after
the 'login' process of selenium has finished. That allows to retrieve informations
that will be needed in the rest of the login process.
After that, the login proceed as normal except for the overriden make_device_print and
make make_transaction_id where we used values directly from selenium.
"""
SELENIUM_BROWSER = AmericanExpressSeleniumFingerprintBrowser
def __init__(self, *args, **kwargs):
super(AmericanExpressWithSeleniumBrowser, self).__init__(*args, **kwargs)
self.selenium_login_transaction_id = None
self.selenium_device_print = None
self.selenium_user_agent = None
self.__states__ += ('selenium_user_agent', )
def do_login(self, *args, **kwargs):
AmericanExpressBrowser.do_login(self, *args, **kwargs)
def load_state(self, *args, **kwargs):
super(AmericanExpressWithSeleniumBrowser, self).load_state(*args, **kwargs)
if self.selenium_user_agent:
self.session.headers['User-Agent'] = self.selenium_user_agent
def load_selenium_session(self, selenium):
self.clear_init_cookies()
super(AmericanExpressWithSeleniumBrowser, self).load_selenium_session(selenium)
# We need to send this value in the login request.
self.selenium_login_transaction_id = selenium.driver.execute_script("return window.inauth._cc[0][1].tid;")
# Save the device print and the user-agent from selenium to replicate the website as much as possible
self.selenium_device_print = selenium.driver.execute_script('return RSA.encode_deviceprint();')
self.selenium_user_agent = selenium.driver.execute_script("return navigator.userAgent;")
self.session.headers['User-Agent'] = self.selenium_user_agent
def setup_browser_for_login_request(self):
SubSeleniumMixin.do_login(self)
def make_device_print(self):
assert self.selenium_device_print
return self.selenium_device_print
def make_transaction_id(self):
assert self.selenium_login_transaction_id
return self.selenium_login_transaction_id
......@@ -21,6 +21,7 @@
import json
from itertools import cycle
from base64 import b64encode, b64decode
import time
from woob.browser.pages import RawPage
......@@ -79,12 +80,12 @@ def get_I(self):
assert match, "Could not find the secret I"
return match[1]
def make_payload_for_s2(self, tid, now):
def make_payload_for_s2(self, tid):
cookie = self.browser.session.cookies['_cc-x']
user_agent = self.browser.session.headers['User-Agent']
return encode(self.make_payload(tid, cookie, user_agent, now))
return encode(self.make_payload(tid, cookie, user_agent))
def make_payload(self, tid, cookie_cc, user_agent, now):
def make_payload(self, tid, cookie_cc, user_agent):
"""
Create a payload for the s2 verification.
The original code in is cc.js where it has the name run.
......@@ -135,9 +136,10 @@ def make_payload(self, tid, cookie_cc, user_agent, now):
That way you can debug without hiting the server.
"""
time_local = now.strftime('%-m/%-d/%Y, %-I:%-M:%-S %p')
time_string = now.strftime('%a %b %d %Y %I:%M:%S GMT+0000 (UTC)')
unix_epoch = round(now.timestamp() * 1000)
unix_epoch = int(time.time()*1000)
time_tuple = time.gmtime()
time_local = time.strftime('%-m/%-d/%Y, %-I:%-M:%-S %p', time_tuple)
time_string = time.strftime('%a %b %d %Y %I:%M:%S GMT+0000 (UTC)', time_tuple)
return {
"sid": "ee490b8fb9a4d570",
......
......@@ -22,7 +22,7 @@
from woob.tools.backend import Module, BackendConfig
from woob.tools.value import ValueBackendPassword, ValueTransient
from .browser import AmericanExpressBrowser
from .browser import AmericanExpressWithSeleniumBrowser
__all__ = ['AmericanExpressModule']
......@@ -41,7 +41,7 @@ class AmericanExpressModule(Module, CapBank):
ValueTransient('request_information'),
ValueTransient('otp', regexp=r'^\d{6}$'),
)
BROWSER = AmericanExpressBrowser
BROWSER = AmericanExpressWithSeleniumBrowser
def create_default_browser(self):
return self.create_browser(
......
......@@ -31,6 +31,7 @@
from woob.capabilities.bank import Account, Transaction
from woob.capabilities.base import NotAvailable
from woob.exceptions import ActionNeeded, BrowserUnavailable
from woob.browser.selenium import SeleniumPage
from dateutil.parser import parse as parse_date
......@@ -234,3 +235,7 @@ def obj_original_amount(self):
return original_amount
obj__ref = Dict('identifier')
class SLoginPage(SeleniumPage):
pass
......@@ -180,7 +180,7 @@ def parse(self, obj):
class AccountHistoryPage(LoggedPage, JsonPage):
def belongs(self, instructions, account):
for ins in instructions:
if 'nomDispositif' in ins and 'codeDispositif' in ins and '%s%s' % (
if ins['type'] != 'ARB' and 'nomDispositif' in ins and 'codeDispositif' in ins and '%s%s' % (
ins['nomDispositif'], ins['codeDispositif']) == '%s%s' % (account.label, account.id):
return True
return False
......
......@@ -29,6 +29,7 @@
from functools import wraps
from dateutil.relativedelta import relativedelta
from requests.exceptions import ReadTimeout
from woob.exceptions import BrowserIncorrectPassword, BrowserUnavailable
from woob.browser.exceptions import HTTPNotFound, ClientError, ServerError
......@@ -50,7 +51,7 @@
NewLoginPage, JsFilePage, AuthorizePage, LoginTokensPage, VkImagePage,
AuthenticationMethodPage, AuthenticationStepPage, CaissedepargneVirtKeyboard,
AccountsNextPage, GenericAccountsPage, InfoTokensPage, NatixisUnavailablePage,
RedirectErrorPage, BPCEPage,
RedirectErrorPage, BPCEPage, AuthorizeErrorPage,
)
from .document_pages import BasicTokenPage, SubscriberPage, SubscriptionsPage, DocumentsPage
from .linebourse_browser import LinebourseAPIBrowser
......@@ -206,6 +207,8 @@ class BanquePopulaire(LoginBrowser):
UnavailablePage
)
authorize_error = URL(r'https://[^/]+/dacswebssoissuer/AuthnRequestServlet', AuthorizeErrorPage)
redirect_page = URL(r'https://[^/]+/portailinternet/_layouts/Ibp.Cyi.Layouts/RedirectSegment.aspx.*', RedirectPage)
bpce_page = URL(r'https://[^/]+/cyber/ibp/ate/portal/internet89C3Portal.jsp', BPCEPage)
......@@ -421,7 +424,15 @@ def do_new_login(self):
'subscribeTypeItems': [],
},
}
self.user_info.go(headers=headers, json=data)
try:
self.user_info.go(headers=headers, json=data)
except ReadTimeout:
# This server usually delivers data in less than a second on this request.
# If it timeouts, retrying will not help.
# It usually comes back up within a few hours.
raise BrowserUnavailable('Le service est momentanément indisponible. Veuillez réessayer plus tard.')
self.user_type = self.page.get_user_type()
user_code = self.page.get_user_code()
......@@ -461,6 +472,8 @@ def do_new_login(self):
self.authorize.go(params=params)
self.page.send_form()
if self.authorize_error.is_here():
raise BrowserUnavailable(self.page.get_error_message())
self.page.check_errors(feature='login')
self.get_current_subbank()
......
......@@ -295,8 +295,18 @@ def is_unavailable(self):
return bool(CleanText('//p[contains(text(), "momentanément indisponible")]')(self.doc))
class AuthorizeErrorPage(HTMLPage):
def is_here(self):
return CleanText('//p[contains(text(), "momentanément indisponible")]')(self.doc)
def get_error_message(self):
return CleanText('//p[contains(text(), "momentanément indisponible")]')(self.doc)
class ErrorPage(LoggedPage, MyHTMLPage):
def on_load(self):
if CleanText('//pre[contains(text(), "unexpected error")]')(self.doc):
raise BrowserUnavailable('An unexpected error has occured.')
if CleanText('//script[contains(text(), "momentanément indisponible")]')(self.doc):
raise BrowserUnavailable("Le service est momentanément indisponible")
elif CleanText('//h1[contains(text(), "Cette page est indisponible")]')(self.doc):
......
......@@ -56,7 +56,7 @@
RecipientsPage, ValidateTransferPage, RegisterTransferPage, AdvisorPage,
AddRecipPage, ActivateRecipPage, ProfilePage, ListDetailCardPage, ListErrorPage,
UselessPage, TransferAssertionError, LoanDetailsPage, TransfersPage, OTPPage,
UnavailablePage, InitLoginPage, FinalizeLoginPage,
UnavailablePage, InitLoginPage, FinalizeLoginPage, InfoClientPage, LoginRedirectPage,
)
from .document_pages import DocumentsPage, TitulairePage, RIBPage
......@@ -71,11 +71,21 @@ class BNPParibasBrowser(LoginBrowser, StatesMixin):
InitLoginPage
)
info_client = URL(
r'/serviceinfosclient-wspl/rpc/InfosClient\?modeAppel=0',
InfoClientPage
)
login = URL(
r'https://connexion-mabanque.bnpparibas/login',
LoginPage
)
login_redirect = URL(
r'https://.*/fr/connexion\?',
LoginRedirectPage,
)
finalize_login = URL(
r'SEEA-pa01/devServer/seeaserver',
FinalizeLoginPage
......@@ -90,8 +100,8 @@ class BNPParibasBrowser(LoginBrowser, StatesMixin):
)
useless_page = URL(
r'/fr/connexion/comptes-et-contrats',
r'/fr/secure/comptes-et-contrats',
r'https://.*/fr/secure/comptes-et-contrats',
r'https://.*/fr/connexion/comptes-et-contrats',
UselessPage
)
......@@ -178,30 +188,40 @@ def do_login(self):
if not (self.username.isdigit() and self.password.isdigit()):
raise BrowserIncorrectPassword()
try:
self.init_login.go(
params={
'client_id': '0e0fe16f-4e44-4138-9c46-fdf077d56087',
'scope': 'openid bnpp_mabanque ikpi',
'response_type': 'code',
'redirect_uri': 'https://mabanque.bnpparibas/fr/connexion',
'ui': 'classic part',
'ui_locales': 'fr',
'wcm_referer': 'mabanque.bnpparibas/',
}
)
self.page.login(self.username, self.password)
except ClientError as e:
# We have to call the page manually with the response
# in order to get the error message
message = LoginPage(self, e.response).get_error()
# Get dynamically error messages
rep = self.errors_list.open()
error_message = rep.json().get(message).replace('<br>', ' ')
exception = self.get_exception_from_message(message, error_message)
raise exception
self.info_client.go()
assert self.info_client.is_here()
if self.page.logged:
# Nothing to do as we are still logged in
return
self.init_login.go(
params={
'client_id': '0e0fe16f-4e44-4138-9c46-fdf077d56087',
'scope': 'openid bnpp_mabanque ikpi',
'response_type': 'code',
'redirect_uri': 'https://mabanque.bnpparibas/fr/connexion',
'ui': 'classic part',
'ui_locales': 'fr',
'wcm_referer': 'mabanque.bnpparibas/',
}
)
# Sometimes, if the session cookie is still valid, the login step is skipped
if self.login.is_here():
try:
self.page.login(self.username, self.password)
except ClientError as e:
# We have to call the page manually with the response
# in order to get the error message
message = LoginPage(self, e.response).get_error()
# Get dynamically error messages
rep = self.errors_list.open()
error_message = rep.json().get(message, '').replace('<br>', ' ')
exception = self.get_exception_from_message(message, error_message)
raise exception
assert self.login_redirect.is_here(), "Not on the authorization redirection page"
code = QueryValue(None, 'code').filter(self.url)
auth = (
......@@ -235,7 +255,7 @@ def do_login(self):
# For some errors, bnporc doesn't return a 403 but redirect to the login page with an error message
# Instead of following the redirection, we parse the errorCode and raise exception with accurate error message
error_code = QueryValue(None, 'errorCode', default=None).filter(
self.response.headers.get('location')
self.response.headers.get('location', '')
)
if error_code:
self.list_error_page.go()
......@@ -263,6 +283,7 @@ def get_exception_from_message(self, message, error_message):
},
BrowserUnavailable: {
'authenticationFailure.TechnicalException900',
'authenticationFailure.TechnicalException917',
'authenticationFailure.TechnicalException901',
'authenticationFailure.TechnicalException902',
'authenticationFailure.TechnicalException903',
......@@ -375,24 +396,35 @@ def iter_accounts(self):
self.accounts_list.append(account)
# Fetching capitalisation contracts from the "Assurances Vie" space (some are not in the BNP API):
params = self.natio_vie_pro.go().get_params()
try:
# When the space does not exist we land on a 302 that tries to redirect
# to an unexisting domain, hence the 'allow_redirects=False'.
# Sometimes the Life Insurance space is unavailable, hence the 'ConnectionError'.
self.location(self.capitalisation_page.build(params=params), allow_redirects=False)
except (ServerError, ConnectionError):
self.logger.warning("An Internal Server Error occurred")
self.natio_vie_pro.go()
message = self.page.get_life_insurance_unavailable_message()
# It seems that natio_vie_pro can return an error message and from that we are not able to make
# requests on the natio insurance life space.
if message:
# "Probleme lors du cryptage des DAT" is the main error returned
# To keep under watch if there is changes about this spaces
self.logger.warning("Natio life insurance space is unavailable : " + message)
else:
if self.capitalisation_page.is_here() and self.page.has_contracts():
for account in self.page.iter_capitalisation():
# Life Insurance accounts may appear BOTH in the API and the "Assurances Vie" domain,
# It is better to keep the API version since it contains the unitvalue:
if account.number not in [a.number for a in self.accounts_list]:
self.logger.warning("We found an account that only appears on the old BNP website.")
self.accounts_list.append(account)
else:
self.logger.warning("This account was skipped because it already appears in the API.")
params = self.page.get_params()
try:
# When the space does not exist we land on a 302 that tries to redirect
# to an unexisting domain, hence the 'allow_redirects=False'.
# Sometimes the Life Insurance space is unavailable, hence the 'ConnectionError'.
self.location(self.capitalisation_page.build(params=params), allow_redirects=False)
except (ServerError, ConnectionError):
self.logger.warning("An Internal Server Error occurred")
else:
if self.capitalisation_page.is_here() and self.page.has_contracts():
for account in self.page.iter_capitalisation():
# Life Insurance accounts may appear BOTH in the API and the "Assurances Vie" domain,
# It is better to keep the API version since it contains the unitvalue:
if account.number not in [a.number for a in self.accounts_list]:
self.logger.warning("We found an account that only appears on the old BNP website.")
self.accounts_list.append(account)
else:
self.logger.warning("This account was skipped because it already appears in the API.")
return iter(self.accounts_list)
......
......@@ -39,7 +39,7 @@
Account, Recipient, Transfer, TransferBankError,
AddRecipientBankError, AccountOwnership,
Emitter, EmitterNumberType, TransferStatus,
TransferDateType,
TransferDateType, TransferInvalidAmount,
)
from woob.capabilities.wealth import (
Investment, MarketOrder, MarketOrderDirection,
......@@ -49,6 +49,7 @@
from woob.capabilities.profile import Person, ProfileMissing
from woob.exceptions import (
BrowserUnavailable, AppValidationCancelled, AppValidationExpired,
AuthMethodNotImplemented,
)
from woob.tools.capabilities.bank.iban import rib2iban, rebuild_rib, is_iban_valid
from woob.tools.capabilities.bank.transactions import FrenchTransaction, parse_with_patterns
......@@ -139,6 +140,10 @@ def get_error(self):
)(self.doc)
class LoginRedirectPage(RawPage):
pass
class FinalizeLoginPage(RawPage):
pass
......@@ -147,6 +152,13 @@ class OTPPage(HTMLPage):
pass
class InfoClientPage(JsonPage):
@property
def logged(self):
message = Dict('message')(self.doc)
return message == 'OK'
class BNPPage(LoggedPage, JsonPage):
def build_doc(self, text):
try:
......@@ -423,7 +435,7 @@ def has_digital_key(self):
class ValidateTransferPage(BNPPage):
def check_errors(self):
if 'data' not in self.doc:
if 'data' not in self.doc or self.doc['message'] != 'OK':
raise TransferBankError(message=self.doc['message'])
def abort_if_unknown(self, transfer_data):
......@@ -477,14 +489,45 @@ def handle_response(self, account, recipient, amount, reason):
class RegisterTransferPage(ValidateTransferPage):
def check_af_validation(self, transfer_data):
sms_id = transfer_data.get('idTransactionSMS')
if sms_id:
raise AuthMethodNotImplemented("La validation des virements par authentification SMS n'est pas supportée.")
app_id = transfer_data.get('idTransactionAF')
if app_id:
raise AuthMethodNotImplemented("La validation des virements par authentification clé digitale n'est pas supportée.")
def handle_response(self, transfer):
self.check_errors()
transfer_data = self.doc['data']['enregistrementVirement']
self.check_af_validation(transfer_data)
transfer.id = transfer_data['reference']
transfer.exec_date = parse_french_date(self.doc['data']['enregistrementVirement']['dateExecution']).date()
plafond_error = transfer_data['montantPlafond']
cumul_error = transfer_data['montantCumule']
reference = transfer_data['reference']
type_operation = transfer_data.get('typeOperation', '')
if plafond_error:
raise TransferInvalidAmount(message="Le montant du virement dépasse le plafond autorisé")
if cumul_error:
raise TransferInvalidAmount(message="Le montant cumulé des virements effectués aujourd'hui dépasse la limite quotidienne autorisée")
if type_operation == 'MAIL' or 'MAIL' in reference:
raise TransferBankError(message="Les caractéristiques de cette opération ne permettent pas sa réalisation. Veuillez contacter votre agence")
# In theory, type operation should be one of:
# "1" - Immediat transfer (ie instant and first open day)
# "2" - Scheduled
# The transfer initation is not registered/executed if any other value
assert type_operation in ['1', '2'], 'Transfer operation type is %s' % type_operation
transfer.id = reference
transfer.exec_date = parse_french_date(transfer_data['dateExecution']).date()
# Timestamp at which the bank registered the transfer
register_date = re.sub(' 24:', ' 00:', self.doc['data']['enregistrementVirement']['dateEnregistrement'])
register_date = re.sub(' 24:', ' 00:', transfer_data['dateEnregistrement'])
transfer._register_date = parse_french_date(register_date)
return transfer
......@@ -687,6 +730,9 @@ class LifeInsurancesDetailPage(LifeInsurancesPage):
class NatioVieProPage(BNPPage):
def get_life_insurance_unavailable_message(self):
return self.doc.get('message')
# This form is required to go to the capitalisation contracts page.
def get_params(self):
params = {
......
......@@ -21,27 +21,53 @@
from datetime import timedelta, datetime
from woob.browser import LoginBrowser, need_login, URL
from woob.browser import LoginBrowser, need_login, URL, StatesMixin
from woob.exceptions import BrowserQuestion, NeedInteractiveFor2FA
from woob.capabilities.bill import DocumentTypes, Document
from woob.tools.capabilities.bank.investments import create_french_liquidity
from woob.tools.value import Value
from .pages import (
LoginPage, HomeLendPage, PortfolioPage, OperationsPage,
LoginPage, OtpPage,
HomeLendPage, PortfolioPage, OperationsPage,
MAIN_ID, ProfilePage, MainPage,
)
class BoldenBrowser(LoginBrowser):
class BoldenBrowser(LoginBrowser, StatesMixin):
BASEURL = 'https://bolden.fr/'
main = URL(r'$', MainPage)
login = URL(r'/connexion', LoginPage)
otp = URL(r'/Account/VerifyCode', OtpPage)
home_lend = URL(r'/tableau-de-bord-investisseur', HomeLendPage)
profile = URL(r'/mon-profil', ProfilePage)
portfolio = URL(r'/InvestorDashboard/GetPortfolio', PortfolioPage)
operations = URL(r'/InvestorDashboard/GetOperations\?startDate=(?P<start>[\d-]+)&endDate=(?P<end>[\d-]+)', OperationsPage)
def __init__(self, config, *args, **kwargs):
kwargs['username'] = config['login'].get()
kwargs['password'] = config['password'].get()
super(BoldenBrowser, self).__init__(*args, **kwargs)
self.config = config
# else the otp page message is in English
self.session.headers['Accept-Language'] = 'fr,fr-FR'
def do_login(self):
def try_otp():
if self.config['otp'].get():
self.page.send_otp(self.config['otp'].get())
assert self.home_lend.is_here()
return True
if self.otp.is_here():
if try_otp():
return
else:
raise NeedInteractiveFor2FA()
self.main.go()
self.page.check_website_maintenance()
self.login.go()
......@@ -51,6 +77,14 @@ def do_login(self):
self.page.check_error()
raise AssertionError('Should not be on login page.')
if self.otp.is_here():
if not try_otp():
if self.config['request_information'] is None:
raise NeedInteractiveFor2FA()
message = self.page.get_otp_message()
raise BrowserQuestion(Value('otp', label=message))
@need_login
def iter_accounts(self):
self.portfolio.go()
......
......@@ -20,7 +20,7 @@
from __future__ import unicode_literals
from woob.tools.backend import Module, BackendConfig
from woob.tools.value import ValueBackendPassword
from woob.tools.value import ValueBackendPassword, ValueTransient
from woob.capabilities.bank import Account
from woob.capabilities.wealth import CapBankWealth
from woob.capabilities.base import find_object
......@@ -49,12 +49,14 @@ class BoldenModule(Module, CapBankWealth, CapDocument, CapProfile):
CONFIG = BackendConfig(
ValueBackendPassword('login', label='Email', masked=False),
ValueBackendPassword('password', label='Mot de passe'),
ValueTransient('otp'),
ValueTransient('request_information'),
)
accepted_document_types = (DocumentTypes.OTHER,)
def create_default_browser(self):
return self.create_browser(self.config['login'].get(), self.config['password'].get())
return self.create_browser(self.config)
def iter_accounts(self):
return self.browser.iter_accounts()
......
......@@ -69,6 +69,17 @@ def check_error(self):
raise ActionNeeded(message)
class OtpPage(HTMLPage):
def send_otp(self, otp):
form = self.get_form(xpath='//form[contains(@action, "Verify")]')
form['Code'] = otp
form['RememberMe'] = 'true'
form.submit()
def get_otp_message(self):
return CleanText('//div[p[contains(text(), "code de vérification")]]')(self.doc)
class HomeLendPage(LoggedPage, HTMLPage):
pass
......
......@@ -192,10 +192,29 @@ def iter_investment(self):
inv.quantity = CleanDecimal.French().filter(info[2])
inv.original_currency = Currency().filter(info[4])
# we need to check if the investment's currency is GBX
# GBX is not part of the ISO4217, to handle it, we need to hardcode it
# first, we check there is a currency string after the unitvalue
unitvalue_currency = info[4].split()
if len(unitvalue_currency) > 1:
# we retrieve the currency string
currency = unitvalue_currency[1]
# we check if the currency notation match the Penny Sterling(GBX)
# example : 1234,5 p
if currency == 'p':
inv.original_currency = 'GBP'
# if not, we can use the regular Currency filter
else:
inv.original_currency = Currency().filter(info[4])
# info[4] = '123,45 &euro;' for investments made in euro, so this filter will return None
if inv.original_currency:
inv.original_unitvalue = CleanDecimal.French().filter(info[4])
# if the currency string is Penny Sterling
# we need to adjust the unitvalue to convert it to GBP
if currency == 'p':
inv.original_unitvalue = CleanDecimal.French().filter(info[4]) / 100
else:
inv.original_unitvalue = CleanDecimal.French().filter(info[4])
else:
# info[4] may be empty so we must handle the default value
inv.unitvalue = CleanDecimal.French(default=NotAvailable).filter(info[4])
......
......@@ -104,20 +104,20 @@ def parse(self, el):
else:
self.env['category'] = 'Externe'
_id = CleanText(Attr('.', 'value'))(el)
if _id == self.env['account_id']:
raise SkipItem()
self.env['id'] = _id
if self.env['category'] == 'Interne':
_id = CleanText(Attr('.', 'value'))(el)
if _id == self.env['account_id']:
raise SkipItem()
try:
account = find_object(self.page.browser.get_accounts_list(), id=_id, error=AccountNotFound)
self.env['id'] = _id
self.env['label'] = account.label
self.env['iban'] = account.iban
except AccountNotFound:
# Some internal recipients cannot be parsed on the website and so, do not
# appear in the iter_accounts. We can still do transfer to those accounts
# because they have an internal id (internal id = id that is not an iban).
self.env['id'] = _id
self.env['iban'] = NotAvailable
raw_label = CleanText('.')(el)
if '-' in raw_label:
......@@ -134,14 +134,21 @@ def parse(self, el):
self.env['bank_name'] = 'La Banque Postale'
else:
self.env['id'] = self.env['iban'] = Regexp(CleanText('.'), '- (.*?) -')(el).replace(' ', '')
self.env['label'] = Regexp(CleanText('.'), '- (.*?) - (.*)', template='\\2')(el).strip()
first_part = CleanText('.')(el).split('-')[0].strip()
if first_part in ['CCP', 'PEL']:
self.env['bank_name'] = 'La Banque Postale'
self.env['iban'] = _id
raw_label = CleanText('.')(el).strip()
# Normally, a beneficiary label looks like that:
# <option value="FR932004...3817">CCP - FR 93 2004...38 17 - MR JOHN DOE</option>
# but sometimes, the label is short, as it can be customized by customers:
# <option value="FR932004...3817">JOHNJOHN</option>
self.env['bank_name'] = NotAvailable
label_parts = raw_label.split(' - ', 2)
if len(label_parts) == 3 and label_parts[1].replace(' ', '') == _id:
if label_parts[0].strip() in ['CCP', 'PEL', 'LJ', 'CEL', 'LDDS']:
self.env['bank_name'] = 'La Banque Postale'
self.env['label'] = label_parts[2].strip()
else:
self.env['bank_name'] = NotAvailable
self.env['label'] = raw_label
if self.env['id'] in self.parent.objects: # user add two recipients with same iban...
raise SkipItem()
......
......@@ -387,8 +387,7 @@ def on_load(self):
page = self.browser.open('/particuliers/compte-bancaire/comptes-en-ligne/bredconnect-compte-ligne?errorCode=%s' % code).page
msg = CleanText('//label[contains(@class, "error")]', default=None)(page.doc)
# 20100: invalid login/password
# 139: dispobank user trying to connect to Bred
if code in ('20100', '139'):
if code == '20100':
raise BrowserIncorrectPassword(msg)
# 20104 & 1000: unknown error during login
elif code in ('20104', '1000'):
......
from .browser import DispoBankBrowser
__all__ = ['DispoBankBrowser']
# -*- coding: utf-8 -*-
# Copyright(C) 2012-2014 Romain Bignon
#
# This file is part of a woob module.
#
# This woob module is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This woob module is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this woob module. If not, see <http://www.gnu.org/licenses/>.
from woob.browser import LoginBrowser, need_login, URL
from woob.exceptions import BrowserIncorrectPassword
from .pages import LoginPage, LoginResultPage, AccountsPage, EmptyPage, TransactionsPage
__all__ = ['DispoBankBrowser']
class DispoBankBrowser(LoginBrowser):
BASEURL = 'https://www.dispobank.fr'
login_page = URL(r'https://www.\w+.fr/mylittleform.*', LoginPage)
login_result = URL(r'https://www.\w+.fr/Andromede/MainAuth.*', LoginResultPage)
accounts_page = URL(r'https://www.\w+.fr/Andromede/Main', AccountsPage)
transactions_page = URL(r'https://www.\w+.fr/Andromede/Ecriture', TransactionsPage)
empty_page = URL(r'https://www.\w+.fr/Andromede/applications/index.jsp',
r'https://www.bred.fr/',
EmptyPage)
login2 = URL(r'https://www.dispobank.fr/?', LoginPage)
URLS = {'bred': {'home': 'https://www.bred.fr/Andromede/Main',
'login': 'https://www.bred.fr/mylittleform?type=1',
},
'dispobank': {'home': 'https://www.dispobank.fr',
'login': 'https://www.dispobank.fr',
}
}
def __init__(self, accnum, config, *args, **kwargs):
login = config['login'].get()
password = config['password'].get()
super(DispoBankBrowser, self).__init__(login, password, *args, **kwargs)
self.accnum = accnum
self.website = 'dispobank'
def home(self):
self.location(self.URLS[self.website]['home'])
def do_login(self):
if not (self.login_page.is_here() or self.login2.is_here()):
self.location(self.URLS[self.website]['login'])
self.page.login(self.username, self.password)
assert self.login_result.is_here() or self.empty_page.is_here()
if self.login_result.is_here():
error = self.page.get_error()
if error is not None:
raise BrowserIncorrectPassword(error)
self.page.confirm()
@need_login
def get_accounts_list(self):
if not self.accounts_page.is_here():
self.location('https://www.%s.fr/Andromede/Main' % self.website)
return self.page.get_list()
@need_login
def get_history(self, account, coming=False):
if coming:
raise NotImplementedError()
numero_compte, numero_poste = account.id.split('.')
data = {'typeDemande': 'recherche',
'motRecherche': '',
'numero_compte': numero_compte,
'numero_poste': numero_poste,
'detail': '',
'tri': 'date',
'sens': 'sort',
'monnaie': 'EUR',
'index_hist': 4
}
self.location('https://www.%s.fr/Andromede/Ecriture' % self.website, data=data)
assert self.transactions_page.is_here()
return self.page.get_history()
def iter_investments(self, account):
raise NotImplementedError()
def iter_market_orders(self, account):
raise NotImplementedError()
@need_login
def get_profile(self):
if not self.accounts_page.is_here():
self.location('https://www.%s.fr/Andromede/Main' % self.website)
return self.page.get_profile()
# -*- coding: utf-8 -*-
# Copyright(C) 2012 Romain Bignon
#
# This file is part of a woob module.
#
# This woob module is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This woob module is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this woob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from decimal import Decimal, InvalidOperation
import re
from collections import OrderedDict
from woob.browser.pages import LoggedPage, HTMLPage, RawPage, FormNotFound
from woob.browser.filters.standard import CleanText
from woob.tools.misc import to_unicode
from woob.capabilities.bank import Account
from woob.capabilities.profile import Profile
from woob.tools.capabilities.bank.transactions import FrenchTransaction
class LoginPage(HTMLPage):
def login(self, login, passwd):
try:
length = int(self.doc.xpath('//input[@id="pass"]')[0].attrib['maxlength'])
except (IndexError,KeyError):
pass
else:
passwd = passwd[:length]
form = self.get_form(name='authen')
form['id'] = login.encode(self.encoding)
form['pass'] = passwd.encode(self.encoding)
form.submit()
class LoginResultPage(HTMLPage):
def on_load(self):
for script in self.doc.xpath('//script'):
text = script.text
if text is None:
continue
m = re.search("window.location.replace\('([^']+)'\);", text)
if m:
self.browser.location(m.group(1))
try:
self.get_form(name='banque')
except FormNotFound:
pass
else:
self.browser.set_all_readonly(False)
accounts = OrderedDict()
for tr in self.doc.xpath('//table[has-class("compteTable")]/tbody/tr'):
if len(tr.findall('td')) == 0:
continue
attr = tr.xpath('.//a')[0].attrib.get('onclick', '')
m = re.search("value = '(\w+)';(checkAndSubmit\('\w+','(\w+)','(\w+)'\))?", attr)
if m:
typeCompte = m.group(1)
tagName = m.group(3)
if tagName is not None:
value = self.doc.xpath('//input[@name="%s"]' % m.group(3))[int(m.group(4))].attrib['value']
else:
value = typeCompte
accounts[value] = (typeCompte, tagName)
try:
typeCompte, tagName = accounts[self.browser.accnum]
value = self.browser.accnum
except KeyError:
accnums = ', '.join(accounts.keys())
if self.browser.accnum != '00000000000':
self.logger.warning(u'Unable to find account "%s". Available ones: %s' % (self.browser.accnum, accnums))
elif len(accounts) > 1:
self.logger.warning('There are several accounts, please use "accnum" backend parameter to force the one to use (%s)' % accnums)
value, (typeCompte, tagName) = accounts.popitem(last=False)
self.browser['typeCompte'] = typeCompte
if tagName is not None:
self.browser[tagName] = [value]
self.browser.submit()
def confirm(self):
self.browser.location('MainAuth?typeDemande=AC', no_login=True)
def get_error(self):
error = self.doc.xpath('//td[@class="txt_norm2"]')
if len(error) == 0:
return None
error = error[0]
if error.find('b') is not None:
error = error.find('b')
return error.text.strip()
class EmptyPage(LoggedPage, RawPage):
pass
class BredBasePage(LoggedPage, HTMLPage):
def js2args(self, s):
cur_arg = None
args = {}
# For example:
# javascript:reloadApplication('nom_application', 'compte_telechargement', 'numero_poste', '000', 'numero_compte', '12345678901','monnaie','EUR');
for sub in re.findall("'([^']+)'", s):
if cur_arg is None:
cur_arg = sub
else:
args[cur_arg] = sub
cur_arg = None
return args
class AccountsPage(BredBasePage):
ACCOUNT_TYPES = {
'Compte à vue': Account.TYPE_CHECKING,
}
def get_list(self):
for tr in self.doc.xpath('//table[@class="compteTable"]/tr'):
if not tr.attrib.get('class', '').startswith('ligne_'):
continue
cols = tr.findall('td')
if len(cols) < 2:
continue
try:
amount = sum([Decimal(FrenchTransaction.clean_amount(txt)) for txt in cols[-1].itertext() if len(txt.strip()) > 0])
except InvalidOperation:
continue
a = cols[0].find('a')
if a is None:
for a in cols[0].xpath('.//li/a'):
args = self.js2args(a.attrib['href'])
if 'numero_compte' not in args or 'numero_poste' not in args:
self.logger.warning('Card link with strange args: %s' % args)
continue
account = Account()
account.id = '%s.%s' % (args['numero_compte'], args['numero_poste'])
account.label = u'Carte %s' % CleanText().filter(a)
account.balance = amount
account.type = account.TYPE_CARD
account.currency = [account.get_currency(txt) for txt in cols[-1].itertext() if len(txt.strip()) > 0][0]
yield account
continue
args = self.js2args(a.attrib['href'])
if 'numero_compte' not in args or 'numero_poste' not in args:
self.logger.warning('Account link for %r with strange args: %s' % (a.attrib.get('alt', a.text), args))
continue
account = Account()
account.id = u'%s.%s' % (args['numero_compte'], args['numero_poste'])
account.label = to_unicode(a.attrib.get('alt', a.text.strip()))
account.balance = amount
account.currency = [account.get_currency(txt) for txt in cols[-1].itertext() if len(txt.strip()) > 0][0]
account.type = self.ACCOUNT_TYPES.get(account.label, Account.TYPE_UNKNOWN)
yield account
def get_profile(self):
profile = Profile()
text = CleanText('//span[@id="intituleAuth"]')(self.doc)
name_re = re.search(r'M(ME|R|LE) (.*)', text)
profile.name = name_re.group(2)
return profile
class Transaction(FrenchTransaction):
PATTERNS = [(re.compile('^RETRAIT G.A.B. \d+ (?P<text>.*?)( CARTE .*)? LE (?P<dd>\d{2})/(?P<mm>\d{2})/(?P<yy>\d{2}).*'),
FrenchTransaction.TYPE_WITHDRAWAL),
(re.compile('^VIR(EMENT)? (?P<text>.*)'), FrenchTransaction.TYPE_TRANSFER),
(re.compile('^PRLV (?P<text>.*)'), FrenchTransaction.TYPE_ORDER),
(re.compile('^(?P<text>.*) TRANSACTION( CARTE .*)? LE (?P<dd>\d{2})/(?P<mm>\d{2})/(?P<yy>\d{2}) ?(.*)$'),
FrenchTransaction.TYPE_CARD),
(re.compile('^CHEQUE.*'), FrenchTransaction.TYPE_CHECK),
(re.compile('^(CONVENTION \d+ )?COTISATION (?P<text>.*)'),
FrenchTransaction.TYPE_BANK),
(re.compile('^REMISE (?P<text>.*)'), FrenchTransaction.TYPE_DEPOSIT),
(re.compile('^(?P<text>.*)( \d+)? QUITTANCE .*'),
FrenchTransaction.TYPE_ORDER),
(re.compile('^CB PAIEM. EN \d+ FOIS \d+ (?P<text>.*?) LE .* LE (?P<dd>\d{2})/(?P<mm>\d{2})/(?P<yy>\d{2})$'),
FrenchTransaction.TYPE_CARD),
(re.compile('^.* LE (?P<dd>\d{2})/(?P<mm>\d{2})/(?P<yy>\d{2})$'),
FrenchTransaction.TYPE_UNKNOWN),
]
class TransactionsPage(LoggedPage, HTMLPage):
def get_history(self):
cleaner = CleanText().filter
for tr in self.doc.xpath('//div[@class="scrollTbody"]/table//tr'):
cols = tr.findall('td')
if len(cols) < 4:
continue
col_label = cols[1]
if col_label.find('a') is not None:
col_label = col_label.find('a')
date = cleaner(cols[0])
label = cleaner(col_label)
t = Transaction()
# an optional tooltip on page contain the second part of the transaction label.
tooltip = self.doc.xpath('//div[@id="tooltip%s"]' % col_label.attrib.get('id', ''))
raw = label
if len(tooltip) > 0:
raw += u' ' + u' '.join([txt.strip() for txt in tooltip[0].itertext()])
raw = re.sub(r'[ ]+', ' ', raw)
t.parse(date, raw)
# as only the first part of label is important to user, if there are no subpart
# taken by FrenchTransaction regexps, reset the label as first part.
if t.label == t.raw:
t.label = label
debit = cleaner(cols[-2])
credit = cleaner(cols[-1])
t.set_amount(credit, debit)
yield t
......@@ -32,7 +32,6 @@
from woob.tools.value import ValueBackendPassword, Value, ValueTransient
from .bred import BredBrowser
from .dispobank import DispoBankBrowser
__all__ = ['BredModule']
......@@ -48,8 +47,6 @@ class BredModule(Module, CapBankWealth, CapProfile, CapBankTransferAddRecipient)
CONFIG = BackendConfig(
ValueBackendPassword('login', label='Identifiant', masked=False, regexp=r'.{1,32}'),
ValueBackendPassword('password', label='Mot de passe'),
Value('website', label="Site d'accès", default='bred',
choices={'bred': 'BRED', 'dispobank': 'DispoBank'}),
Value('accnum', label='Numéro du compte bancaire (optionnel)', default='', masked=False),
ValueTransient('request_information'),
ValueTransient('resume'),
......@@ -57,17 +54,10 @@ class BredModule(Module, CapBankWealth, CapProfile, CapBankTransferAddRecipient)
ValueTransient('otp_app'),
)
BROWSERS = {
'bred': BredBrowser,
'dispobank': DispoBankBrowser,
}
BROWSER = BredBrowser
def get_website(self):
return self.config['website'].get()
def create_default_browser(self):
self.BROWSER = self.BROWSERS[self.get_website()]
return self.create_browser(
self.config['accnum'].get().replace(' ', '').zfill(11),
self.config,
......@@ -96,9 +86,6 @@ def get_profile(self):
return self.browser.get_profile()
def fill_account(self, account, fields):
if self.get_website() != 'bred':
return
self.browser.fill_account(account, fields)
OBJECTS = {
......@@ -106,18 +93,12 @@ def fill_account(self, account, fields):
}
def iter_transfer_recipients(self, account):
if self.get_website() != 'bred':
raise NotImplementedError()
if not isinstance(account, Account):
account = find_object(self.iter_accounts(), id=account)
return self.browser.iter_transfer_recipients(account)
def new_recipient(self, recipient, **params):
if self.get_website() != 'bred':
raise NotImplementedError()
recipient.label = recipient.label[:32].strip()
regex = r'[-a-z0-9A-Z ,.]+'
......@@ -128,9 +109,6 @@ def new_recipient(self, recipient, **params):
return self.browser.new_recipient(recipient, **params)
def init_transfer(self, transfer, **params):
if self.get_website() != 'bred':
raise NotImplementedError()
transfer.label = transfer.label[:140].strip()
regex = r'[-a-z0-9A-Z ,.]+'
......@@ -150,6 +128,4 @@ def init_transfer(self, transfer, **params):
return self.browser.init_transfer(transfer, account, recipient, **params)
def execute_transfer(self, transfer, **params):
if self.get_website() != 'bred':
raise NotImplementedError()
return self.browser.execute_transfer(transfer, **params)