Skip to content
browser.py 61 KiB
Newer Older
Romain Bignon's avatar
Romain Bignon committed
# Copyright(C) 2010-2011 Nicolas Duhamel
Roger Philibert's avatar
Roger Philibert committed
# This file is part of a woob module.
Roger Philibert's avatar
Roger Philibert committed
# 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
Romain Bignon's avatar
Romain Bignon committed
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
Roger Philibert's avatar
Roger Philibert committed
# This woob module is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
Romain Bignon's avatar
Romain Bignon committed
# 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
Roger Philibert's avatar
Roger Philibert committed
# along with this woob module. If not, see <http://www.gnu.org/licenses/>.
ntome's avatar
ntome committed
# flake8: compatible

import time
baptiste delpey's avatar
baptiste delpey committed
from datetime import datetime, timedelta
from urllib.parse import urlsplit, urlunsplit
from dateutil.relativedelta import relativedelta
from requests.exceptions import HTTPError

Damien Mat's avatar
Damien Mat committed
from woob.browser import URL, LoginBrowser, need_login
from woob.browser.adapters import LowSecHTTPAdapter
from woob.browser.browsers import StatesMixin
Damien Mat's avatar
Damien Mat committed
from woob.browser.exceptions import BrowserHTTPNotFound, ServerError
from woob.capabilities.bank import (
    Account, AccountOwnership, AccountOwnerType, AddRecipientBankError,
    AddRecipientStep, Loan, Recipient, RecipientInvalidOTP, TransferBankError,
    TransferInvalidOTP, TransferInvalidRecipient, TransferStep,
    NoAccountsException,
)
from woob.capabilities.base import NotAvailable
from woob.exceptions import (
Damien Mat's avatar
Damien Mat committed
    ActionNeeded, ActionType, AppValidation, AppValidationCancelled, AppValidationExpired,
    BrowserIncorrectPassword, BrowserQuestion, BrowserUnavailable, BrowserUserBanned,
    NeedInteractiveFor2FA,
from woob.tools.decorators import retry
from woob.tools.url import get_url_param
from woob.tools.value import Value
from woob_modules.linebourse.browser import LinebourseAPIBrowser
Damien Mat's avatar
Damien Mat committed
    AccountDesactivate, AccountHistory, AccountList, AccountRIB, Advisor, BadLoginPage,
    CardsJsonDetails, CardsList, CerticodePlusSubmitDevicePage, CheckPassword, CompleteTransfer,
    ConfirmPage, CreateRecipient, DecoupledPage, DownloadPage, Initident, LoginPage,
    Loi6902TransferPage, OtpErrorPage, PersonalLoanRoutagePage, ProSubscriptionPage,
    ProTransferChooseAccounts, RcptSummary, SmsPage, SubscriptionPage, TemporaryPage,
    TransferChooseAccounts, TransferConfirm, TransferSummary, TwoFAPage, UnavailablePage,
    ValidateCountry, Validated2FAPage, ValidateRecipient, repositionnerCheminCourant,
from .pages.accounthistory import (
Damien Mat's avatar
Damien Mat committed
    LifeInsuranceAccessHistory, LifeInsuranceHistory, LifeInsuranceHistoryInv,
    LifeInsuranceInitPage, LifeInsuranceInvest, LifeInsuranceSummary, RetirementHistory,
    SavingAccountSummary,
from .pages.accountlist import (
Damien Mat's avatar
Damien Mat committed
    MarketCheckPage, MarketHomePage, MarketLoginPage, ProfilePage, RevolvingPage,
    UserTransactionIDPage,
from .pages.base import IncludedUnavailablePage, UselessPage
Damien Mat's avatar
Damien Mat committed
from .pages.login import NoTerminalPage, PostLoginPage
from .pages.mandate import MandateAccountsList, MandateLife, MandateMarket, PreMandate, PreMandateBis
Fong NGO's avatar
Fong NGO committed
from .pages.pro import (
Damien Mat's avatar
Damien Mat committed
    Detect2FAPage, DownloadRib, ProAccountHistory, ProAccountsList, RedirectAfterVKPage,
    RedirectPage, RibPage, SwitchQ5CPage,
Fong NGO's avatar
Fong NGO committed
)
__all__ = ['BPBrowser', 'BProBrowser']
class BPBrowser(LoginBrowser, StatesMixin):
    BASEURL = 'https://voscomptesenligne.labanquepostale.fr'
    HTTP_ADAPTER_CLASS = LowSecHTTPAdapter
    STATE_DURATION = 10
    TIMEOUT = 20
    # FIXME beware that '.*' in start of URL() won't match all domains but only under BASEURL

    included_unavailable_page = URL(r'.*', IncludedUnavailablePage)

    login_image = URL(r'.*wsost/OstBrokerWeb/loginform\?imgid=', UselessPage)
    login_page = URL(r'.*wsost/OstBrokerWeb/loginform.*', LoginPage)
ntome's avatar
ntome committed
    repositionner_chemin_courant = URL(
        r'.*authentification/repositionnerCheminCourant-identif.ea',
        repositionnerCheminCourant
    )
    init_ident = URL(r'.*authentification/initialiser-identif.ea', Initident)
ntome's avatar
ntome committed
    check_password = URL(
        r'.*authentification/verifierMotDePasse-identif.ea',
        r'/securite/authentification/verifierPresenceCompteOK-identif.ea',
        r'.*//voscomptes/identification/motdepasse.jsp',
        CheckPassword
    )

    redirect_page = URL(
        r'.*voscomptes/identification/identification.ea.*',
        r'.*voscomptes/synthese/3-synthese.ea',
        RedirectPage
    )

    auth_page = URL(
        r'voscomptes/canalXHTML/securite/gestionAuthentificationForte/init-gestionAuthentificationForte.ea',
        TwoFAPage
    )
    validated_2fa_page = URL(
        r'voscomptes/canalXHTML/securite/gestionAuthentificationForte/../../securite/authentification/retourDSP2-identif.ea',
        r'voscomptes/canalXHTML/securite/authentification/retourDSP2-identif.ea',
        Validated2FAPage
    )
    decoupled_page = URL(
        r'/voscomptes/canalXHTML/securite/gestionAuthentificationForte/validationTerminal-gestionAuthentificationForte.ea',
        DecoupledPage
    )
    sms_page = URL(
        r'/voscomptes/canalXHTML/securite/gestionAuthentificationForte/authenticateCerticode-gestionAuthentificationForte.ea',
        SmsPage
    )
    sms_validation = URL(
        r'/voscomptes/canalXHTML/securite/gestionAuthentificationForte/validation-gestionAuthentificationForte.ea',
        SmsPage
    )
    no_terminal = URL(
        r'/voscomptes/canalXHTML/securite/gestionAuthentificationForte/initCU633-gestionAuthentificationForte.ea',
        NoTerminalPage
    )
ntome's avatar
ntome committed
    par_accounts_checking = URL(
        '/voscomptes/canalXHTML/comptesCommun/synthese_ccp/afficheSyntheseCCP-synthese_ccp.ea',
        AccountList
    )
    par_accounts_savings_and_invests = URL(
        '/voscomptes/canalXHTML/comptesCommun/synthese_ep/afficheSyntheseEP-synthese_ep.ea',
        AccountList
    )
    par_accounts_loan = URL(
        r'/voscomptes/canalXHTML/pret/encours/consulterPrets-encoursPrets.ea',
        r'/voscomptes/canalXHTML/pret/encours/detaillerPretPartenaireListe-encoursPrets.ea',
        r'/voscomptes/canalXHTML/pret/encours/detaillerOffrePretImmoListe-encoursPrets.ea',
        r'/voscomptes/canalXHTML/pret/encours/detaillerOffrePretConsoListe-encoursPrets.ea',
        r'/voscomptes/canalXHTML/pret/encours/rechercherPret-encoursPrets.ea',
        r'/voscomptes/canalXHTML/sso/commun/init-integration.ea\?partenaire=cristalCEC',
        r'https://espaceclientcreditconso.labanquepostale.fr/esd/contratDetail',
nvergnac's avatar
nvergnac committed
        r'https://espaceclientcreditconso(-esd)?.labanquepostale.fr/esd/contratDetail',
ntome's avatar
ntome committed
        AccountList
    )
    temporary_page = URL(
        r'/voscomptes/canalXHTML/securite/authentification/verifierSyndicationParapheur-identif.ea',
        r'/voscomptes/canalXHTML/sso/espaceDigitalLBPF/init-espaceDigitalLBPF.ea',
        r'voscomptes/canalXHTML/pret/encours/preparerRedirectionEsdLbpf-encoursPrets.ea',
        TemporaryPage
    )
    personal_loan_routage_url = URL(
        '/voscomptes/canalXHTML/sso/espaceDigitalLBPF/routage-espaceDigitalLBPF.ea',
        PersonalLoanRoutagePage
    )
    revolving_redirection = URL(
        '/voscomptes/canalXHTML/pret/encours/detaillerCreditRenouvelableListe-encoursPrets.ea',
        RevolvingPage
    )
    revolving_details = URL(
        '/voscomptes/canalXHTML/pret/creditRenouvelable/init-consulterCreditRenouvelable.ea',
        RevolvingPage
ntome's avatar
ntome committed
    )
    accounts_rib = URL(
        r'.*voscomptes/canalXHTML/comptesCommun/imprimerRIB/init-imprimer_rib.ea.*',
        '/voscomptes/canalXHTML/comptesCommun/imprimerRIB/init-selection_rib.ea',
        AccountRIB
    )

    saving_summary = URL(
        r'/voscomptes/canalXHTML/assurance/vie/reafficher-assuranceVie.ea(\?numContrat=(?P<id>\w+))?',
        r'/voscomptes/canalXHTML/assurance/retraiteUCEuro/afficher-assuranceRetraiteUCEuros.ea(\?numContrat=(?P<id>\w+))?',
        r'/voscomptes/canalXHTML/assurance/retraitePoints/reafficher-assuranceRetraitePoints.ea(\?numContrat=(?P<id>\w+))?',
        r'/voscomptes/canalXHTML/assurance/prevoyance/reafficher-assurancePrevoyance.ea(\?numContrat=(?P<id>\w+))?',
        SavingAccountSummary
    )

    lifeinsurance_summary = URL(
        r'/voscomptes/canalXHTML/assurance/vie/syntheseVie-assuranceVie.ea\?numContrat=(?P<id>\w+)',
        LifeInsuranceSummary
    )
ntome's avatar
ntome committed
    lifeinsurance_invest = URL(
        r'/ws_nsi/api/v2/assurance/valorisations/contrats/(?P<contract_id>\d+)/supports\?codeProduit=(?P<product_code>\d+)',
ntome's avatar
ntome committed
        LifeInsuranceInvest
    )
    lifeinsurance_init_page = URL(
        r'/voscomptes/canalXHTML/sso/nsi/init-nsi.ea',
        LifeInsuranceInitPage,
    )
    lifeinsurance_access_page = URL(
        r'/voscomptes/canalXHTML/sso/nsi/routage-nsi.ea',
        LifeInsuranceAccessHistory,
    )
ntome's avatar
ntome committed
    lifeinsurance_history = URL(
        r'/ws_nsi/api/v2/assurance/mouvements/contrats/(?P<contract_id>\d+)/operations/(?P<product_code>\d+)',
ntome's avatar
ntome committed
        LifeInsuranceHistory
    )
    lifeinsurance_hist_inv = URL(
        r'/voscomptes/canalXHTML/assurance/vie/detailMouvement-assuranceVie.ea\?idMouvement=(?P<id>\w+)',
        r'/voscomptes/canalXHTML/assurance/vie/detailMouvementHermesBompard-assuranceVie.ea\?idMouvement=(\w+)',
        LifeInsuranceHistoryInv
    )
    market_home = URL(r'https://labanquepostale.offrebourse.com/fr/\d+/?', MarketHomePage)
    market_login = URL(r'/voscomptes/canalXHTML/bourse/aiguillage/oicFormAutoPost.jsp', MarketLoginPage)
ntome's avatar
ntome committed
    market_check = URL(
        r'/voscomptes/canalXHTML/bourse/aiguillage/lancerBourseEnLigne-connexionBourseEnLigne.ea',
        MarketCheckPage
    )
    useless = URL(r'https://labanquepostale.offrebourse.com/ReroutageSJR', UselessPage)

ntome's avatar
ntome committed
    retirement_hist = URL(
        r'/voscomptes/canalXHTML/assurance/retraitePoints/historiqueRetraitePoint-assuranceRetraitePoints.ea(\?numContrat=(?P<id>\w+))?',
        r'/voscomptes/canalXHTML/assurance/retraiteUCEuro/historiqueMouvements-assuranceRetraiteUCEuros.ea(\?numContrat=(?P<id>\w+))?',
        r'/voscomptes/canalXHTML/assurance/prevoyance/consulterHistorique-assurancePrevoyance.ea(\?numContrat=(?P<id>\w+))?',
        RetirementHistory
    )

    par_account_checking_history = URL(
        r'/voscomptes/canalXHTML/CCP/releves_ccp/init-releve_ccp.ea\?typeRecherche=4&compte.numero=(?P<accountId>.*)',
        r'/voscomptes/canalXHTML/CCP/releves_ccp/menuReleve-releve_ccp.ea\?compte.numero=(?P<accountId>.*)&typeRecherche=5',
ntome's avatar
ntome committed
        r'/voscomptes/canalXHTML/CCP/releves_ccp/afficher-releve_ccp.ea',
        AccountHistory
    )
    single_card_history = URL(
        r'/voscomptes/canalXHTML/CB/releveCB/preparerRecherche-mouvementsCarteDD.ea\?typeListe=(?P<monthIndex>\d+)',
        AccountHistory
    )
    deferred_card_history = URL(
        r'/voscomptes/canalXHTML/CB/releveCB/init-mouvementsCarteDD.ea\?compte.numero=(?P<accountId>\w+)&indexCompte=(?P<cardIndex>\d+)&typeListe=(?P<monthIndex>\d+)',
        AccountHistory
    )
    deferred_card_history_multi = URL(
        r'/voscomptes/canalXHTML/CB/releveCB/preparerRecherche-mouvementsCarteDD.ea\?indexCompte=(?P<accountId>\w+)&indexCarte=(?P<cardIndex>\d+)&typeListe=(?P<monthIndex>\d+)',
        r'/voscomptes/canalXHTML/CB/releveCB/preparerRecherche-mouvementsCarteDD.ea\?compte.numero=(?P<accountId>\w+)&indexCarte=(?P<cardIndex>\d+)&typeListe=(?P<monthIndex>\d+)',
        AccountHistory
    )
    par_account_checking_coming = URL(
        r'/voscomptes/canalXHTML/CCP/releves_ccp_encours/preparerRecherche-releve_ccp_encours.ea\?compte.numero=(?P<accountId>.*)&typeRecherche=1',
        r'/voscomptes/canalXHTML/CB/releveCB/init-mouvementsCarteDD.ea\?compte.numero=(?P<accountId>.*)&typeListe=1&typeRecherche=10',
        r'/voscomptes/canalXHTML/CCP/releves_ccp_encours/preparerRecherche-releve_ccp_encours.ea\?indexCompte',
        r'/voscomptes/canalXHTML/CNE/releveCNE_encours/init-releve_cne_en_cours.ea\?compte.numero',
        r'/voscomptes/canalXHTML/CNE/releveCNE_encours/init-releve_cne_en_cours.ea\?indexCompte=(?P<accountId>.*)&typeRecherche=1&typeMouvements=CNE',
        AccountHistory
    )
    par_account_savings_and_invests_history = URL(
        r'/voscomptes/canalXHTML/CNE/releveCNE/init-releve_cne.ea\?typeRecherche=10&compte.numero=(?P<accountId>.*)',
        r'/voscomptes/canalXHTML/CNE/releveCNE/releveCNE-releve_cne.ea',
        r'/voscomptes/canalXHTML/CNE/releveCNE/frame-releve_cne.ea\?compte.numero=.*&typeRecherche=.*',
        AccountHistory
    )

    cards_list = URL(
        r'/voscomptes/canalXHTML/CB/releveCB/init-mouvementsCarteDD.ea\?compte.numero=(?P<account_id>\w+)$',
        r'.*CB/releveCB/init-mouvementsCarteDD.ea.*',
        CardsList
    )

    cards_json_detail = URL(
        r'/ws_q44/api/v1/cartes\?numCompte=(?P<account_id>\w+)&typeListe=consultation',
        CardsJsonDetails
    )
ntome's avatar
ntome committed
    transfer_choose = URL(
        r'/voscomptes/canalXHTML/virement/mpiaiguillage/init-saisieComptes.ea',
        TransferChooseAccounts
    )
    transfer_complete = URL(
        r'/voscomptes/canalXHTML/virement/mpiaiguillage/soumissionChoixComptes-saisieComptes.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_national/init-creerVirementNational.ea',
        # The two following urls are obtained after a redirection made after a form
        # No parameters or data seem to change that the website go back to the evious folder, using ".."
        # We can't do much since it is finaly handled by the module requests
        r'/voscomptes/canalXHTML/virement/mpiaiguillage/\.\./virementSafran_national/init-creerVirementNational.ea',
        r'/voscomptes/canalXHTML/virement/mpiaiguillage/\.\./virementSafran_sepa/init-creerVirementSepa.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_sepa/init-creerVirementSepa.ea',
        CompleteTransfer
    )
    popup_loi6902_transfer = URL(
        r'/voscomptes/canalXHTML/virement/popinBlocageLoi6902/popinBlocageLoi6902_(?P<popin_suffix>.*).jsp',
        Loi6902TransferPage
ntome's avatar
ntome committed
    )

    # transfer_summary needs to be before transfer_confirm because both
    # have one url in common with different is_here conditions.
    # We need to check the is_here of transfer_summary first to be on the correct page.
    transfer_summary = URL(
        r'/voscomptes/canalXHTML/virement/virementSafran_national/confirmerVirementNational-virementNational.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_pea/confirmerInformations-virementPea.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_sepa/confirmer-creerVirementSepa.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_national/confirmer-creerVirementNational.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_sepa/confirmerInformations-virementSepa.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_sepa/finalisation-creerVirementSepa.ea',
        TransferSummary
    )
ntome's avatar
ntome committed
    transfer_confirm = URL(
        r'/voscomptes/canalXHTML/virement/virementSafran_pea/validerVirementPea-virementPea.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_sepa/valider-creerVirementSepa.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_sepa/valider-virementSepa.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_sepa/confirmerInformations-virementSepa.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_national/valider-creerVirementNational.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_national/validerVirementNational-virementNational.ea',
        # The following url is already used in transfer_summary
        # but we need it to detect the case where the website displays the list of devices
ntome's avatar
ntome committed
        # when a transfer is made with an otp or decoupled
        r'/voscomptes/canalXHTML/virement/virementSafran_sepa/confirmer-creerVirementSepa.ea',
        TransferConfirm
    )

    create_recipient = URL(
        r'/voscomptes/canalXHTML/virement/mpiGestionBeneficiairesVirementsCreationBeneficiaire/init-creationBeneficiaire.ea',
        r'/voscomptes/canalXHTML/virement/virementSafran_commun/.*.ea',
        CreateRecipient
    )
    validate_country = URL(
        r'/voscomptes/canalXHTML/virement/mpiGestionBeneficiairesVirementsCreationBeneficiaire/validationSaisiePaysBeneficiaire-creationBeneficiaire.ea',
        ValidateCountry
    )
    validate_recipient = URL(
        r'/voscomptes/canalXHTML/virement/mpiGestionBeneficiairesVirementsCreationBeneficiaire/valider-creationBeneficiaire.ea',
        ValidateRecipient
    )
    certicode_plus_submit_device = URL(
ntome's avatar
ntome committed
        r'/voscomptes/canalXHTML/securisation/mpin/demandeCreation-securisationMPIN.ea',
        CerticodePlusSubmitDevicePage
ntome's avatar
ntome committed
    )
    rcpt_code = URL(
        r'/voscomptes/canalXHTML/virement/mpiGestionBeneficiairesVirementsCreationBeneficiaire/validerRecapBeneficiaire-creationBeneficiaire.ea',
        ConfirmPage
    )
    otp_benef_transfer_error = URL(
        r'/voscomptes/canalXHTML/securisation/otp/validation-securisationOTP.ea',
ntome's avatar
ntome committed
    rcpt_summary = URL(
        r'/voscomptes/canalXHTML/virement/mpiGestionBeneficiairesVirementsCreationBeneficiaire/finalisation-creationBeneficiaire.ea',
        RcptSummary
    )

    badlogin = URL(
        r'https://transverse.labanquepostale.fr/.*ost/messages\.CVS\.html\?param=0x132120c8.*',  # still valid?
        r'https://transverse.labanquepostale.fr/xo_/messages/message.html\?param=0x132120c8.*',
        BadLoginPage
    )
    disabled_account = URL(
        r'.*ost/messages\.CVS\.html\?param=0x132120cb.*',
        r'.*/message\.html\?param=0x132120c.*',
        r'https://transverse.labanquepostale.fr/xo_/messages/message.html\?param=0x132120cb.*',
        AccountDesactivate
    )

    unavailable = URL(
        r'https?://.*.labanquepostale.fr/delestage.html',
lbellebouche's avatar
lbellebouche committed
        r'https://transverse.labanquepostale.fr/xo_/messages/message.html\?param=.*',
ntome's avatar
ntome committed
        UnavailablePage
    )
    rib_dl = URL(r'.*/voscomptes/rib/init-rib.ea', DownloadRib)
    rib = URL(r'.*/voscomptes/rib/preparerRIB-rib.*', RibPage)
ntome's avatar
ntome committed
    advisor = URL(
        r'/ws_q45/Q45/canalXHTML/commun/authentification/init-identif.ea\?origin=particuliers&codeMedia=0004&entree=HubHome',
        r'/ws_q45/Q45/canalXHTML/desktop/home/init-home.ea',
        Advisor
    )
ntome's avatar
ntome committed
    login_url = 'https://voscomptesenligne.labanquepostale.fr/wsost/OstBrokerWeb/loginform?TAM_OP=login&ERROR_CODE=0x00000000&URL=%2Fvoscomptes%2FcanalXHTML%2Fidentif.ea%3Forigin%3Dparticuliers'
    post_login_page = URL(
        r'https://voscomptesenligne.labanquepostale.fr/wsost/OstBrokerWeb/auth',
        PostLoginPage
    )
    pre_mandate = URL(r'/voscomptes/canalXHTML/sso/commun/init-integration.ea\?partenaire=procapital', PreMandate)
ntome's avatar
ntome committed
    pre_mandate_bis = URL(
        r'https://www.gestion-sous-mandat.labanquepostale-gestionprivee.fr/lbpgp/secure/main.html',
        PreMandateBis
    )
    mandate_accounts_list = URL(
        r'https://www.gestion-sous-mandat.labanquepostale-gestionprivee.fr/lbpgp/secure/accounts_list.html',
        MandateAccountsList
    )
    mandate_market = URL(
        r'https://www.gestion-sous-mandat.labanquepostale-gestionprivee.fr/lbpgp/secure_account/selectedAccountDetail.html',
        MandateMarket
    )
    mandate_life = URL(
        r'https://www.gestion-sous-mandat.labanquepostale-gestionprivee.fr/lbpgp/secure_main/asvContratClient.html',
        r'https://www.gestion-sous-mandat.labanquepostale-gestionprivee.fr/lbpgp/secure_ajax/asvSupportsDetail.html',
        MandateLife
    )

jlouvel's avatar
jlouvel committed
    user_transaction_id = URL(r'/ws_q44/api/userProfile', UserTransactionIDPage)

    profile_page = URL(r'/ws_q44/api/v1/infoclient/donneesClient', ProfilePage)
ntome's avatar
ntome committed

    subscription = URL(
        '/voscomptes/canalXHTML/relevePdf/relevePdf_historique/reinitialiser-historiqueRelevesPDF.ea',
        SubscriptionPage
    )
    subscription_search = URL(
        r'/voscomptes/canalXHTML/relevePdf/relevePdf_historique/form-historiqueRelevesPDF\.ea',
        SubscriptionPage
    )
    download_page = URL(
        r'/voscomptes/canalXHTML/relevePdf/relevePdf_historique/telechargerPDF-historiqueRelevesPDF.ea\?ts=.*&listeRecherche=.*',
        DownloadPage
    )
Baptiste Delpey's avatar
Baptiste Delpey committed
    accounts = None
    __states__ = ('need_reload_state', 'sms_form')
    def __init__(self, config, *args, **kwargs):
        self.config = config
Damien Mat's avatar
Damien Mat committed
        super().__init__(*args, **kwargs)
        self.resume = config['resume'].get()
        self.request_information = config['request_information'].get()

        dirname = self.responses_dirname
        if dirname:
            dirname += '/bourse'
        self.linebourse = LinebourseAPIBrowser(
            'https://labanquepostale.offrebourse.com/',
            logger=self.logger,
            responses_dirname=dirname,
            proxy=self.PROXIES
        )
        self.need_reload_state = None
        self.profile = None  # Not only used for CapProfile, store it to avoid multiplying ProfilePage requests.
baptiste delpey's avatar
baptiste delpey committed

    def load_state(self, state):
        if state.get('url'):
            # We don't want to come back to last URL during SCA
            state.pop('url')

        if state.get('need_reload_state'):
Damien Mat's avatar
Damien Mat committed
            super().load_state(state)
            self.need_reload_state = None
    def deinit(self):
Damien Mat's avatar
Damien Mat committed
        super().deinit()
        self.linebourse.deinit()

    def open(self, *args, **kwargs):
Damien Mat's avatar
Damien Mat committed
            return super().open(*args, **kwargs)
        except ServerError as err:
            if "/../" not in err.response.url:
                raise
            # this shit website includes ".." in an absolute url in the Location header
            # requests passes it verbatim, and the site can't handle it
            self.logger.debug('site has "/../" in their url, fixing url manually')
            parts = list(urlsplit(err.response.url))
            parts[2] = os.path.abspath(parts[2])
            return self.open(urlunsplit(parts))
Tom LARGE's avatar
Tom LARGE committed
    def try_go_get_subscription_search(self, params):
        try:
            self.subscription_search.go(params=params)
        except ServerError as e:
            if e.response.status_code == 500 and params['formulaire.anneeRecherche'] == '2013':
                # No documents for year 2013. It will return a 500 error.
                return True
            raise
        return False

    def login_without_2fa(self):
        self.location(self.login_url)
        self.page.login(self.username, self.password)
        if self.post_login_page.is_here():
            error_message = self.page.get_error_message()
            if error_message:
                if 'Une erreur est survenue' in error_message:
                    raise BrowserUnavailable(error_message)

                raise AssertionError(f'Error not handled during login: {error_message}')

            raise AssertionError('Unexpected error during login.')

        if self.redirect_page.is_here() and not self.page.is_logged():
            if self.page.check_for_perso():
                raise BrowserIncorrectPassword("L'identifiant utilisé est celui d'un compte de Particuliers.")
            error = self.page.get_error()
            raise BrowserUnavailable(error or '')
        if self.badlogin.is_here():
            raise BrowserIncorrectPassword()
        if self.disabled_account.is_here():
    def do_login(self):
        self.code = self.config['code'].get()
        if self.resume:
            return self.handle_polling()
        elif self.code:
            return self.handle_sms()

        self.login_without_2fa()

        try:
            self.auth_page.go()
        except BrowserHTTPNotFound:
            # Instability of the website. We can try do_login again without 2fa request
            self.login_without_2fa()

        if self.auth_page.is_here():
            auth_method = self.page.get_auth_method()
            if self.request_information is None and auth_method != 'no2fa':
                # We don't want to raise this exception if 2FA is absent
                raise NeedInteractiveFor2FA()

            if auth_method == 'cer+':
                # We force here the first device present
                self.decoupled_page.go(params={'deviceSelected': '0'})
                if self.no_terminal.is_here() and self.page.has_no_terminal():
                    raise ActionNeeded(
                        locale="fr-FR", message="Veuillez associer votre téléphone à votre compte bancaire pour réaliser l'authentification forte",
                        action_type=ActionType.ENABLE_MFA,
                    )
                raise AppValidation(self.page.get_decoupled_message())

            elif auth_method == 'cer':
                self.location('/voscomptes/canalXHTML/securite/gestionAuthentificationForte/authenticateCerticode-gestionAuthentificationForte.ea')
                self.page.check_if_is_blocked()
                self.sms_form = self.page.get_sms_form()
                self.need_reload_state = True
                raise BrowserQuestion(Value('code', label='Entrez le code reçu par SMS'))

Maxime Gasselin's avatar
Maxime Gasselin committed
            elif auth_method == 'no2fa':
                raise ActionNeeded(
                    locale="fr-FR", message="Veuillez activer votre service gratuit d'authentification forte sur votre site bancaire.",
                    action_type=ActionType.ENABLE_MFA,
                )
        # If we are here, we don't need 2FA, we are logged

    def do_polling(self, polling_url):
        timeout = time.time() + 300.00
        while time.time() < timeout:
            polling = self.location(polling_url, allow_redirects=False)
            if polling.status_code == 302:
                # Session expired.
                raise AppValidationExpired()

            result = polling.json()['statutOperation']
            if result == '1':
                # Waiting for PSU validation
                continue
            elif result == '2':
                # Validated
                break
            elif result == '3':
                raise AppValidationCancelled()
            elif result == '6' or result == 'ERR':
                raise AppValidationExpired()
            else:
ntome's avatar
ntome committed
                raise AssertionError('statutOperation: %s is not handled' % result)
            raise AppValidationExpired()

    def handle_polling(self):
        polling_url = self.absurl(
            '/voscomptes/canalXHTML/securite/gestionAuthentificationForte/validationOperation-gestionAuthentificationForte.ea',
            base=True
        )
        self.do_polling(polling_url=polling_url)
        self.location(self.absurl(
            '/voscomptes/canalXHTML/securite/gestionAuthentificationForte/finalisation-gestionAuthentificationForte.ea',
            base=True
        ))

    def handle_sms(self):
        self.sms_form['codeOTPSaisi'] = self.code
        self.sms_validation.go(data=self.sms_form)
        if not self.validated_2fa_page.is_here() and self.page.is_sms_wrong():
            raise BrowserIncorrectPassword('Le code de sécurité que vous avez saisi est erroné.')

    @need_login
    def get_accounts_list(self):
        if self.session.cookies.get('indicateur'):
            # Malformed cookie to delete to reach other spaces
            del self.session.cookies['indicateur']

Baptiste Delpey's avatar
Baptiste Delpey committed
        if self.accounts is None:
            accounts = []
            to_check = []
            mandate_urls = []
            self.par_accounts_checking.go()
            pages = [self.par_accounts_checking, self.par_accounts_savings_and_invests, self.par_accounts_loan]
            no_accounts = 0
            for page in pages:
                page.go()
                assert page.is_here(), "AccountList type page not reached"
                if self.page.no_accounts:
                    no_accounts += 1
                    continue
                mandate_urls.extend(self.page.get_mandate_accounts_urls())

                for account in self.page.iter_accounts():
Heddy's avatar
Heddy committed
                    if account.type in (Account.TYPE_LOAN, Account.TYPE_MORTGAGE):
                        accounts.extend(self.get_loans(account))
                    elif account.type == Account.TYPE_LIFE_INSURANCE:
                        self.lifeinsurance_summary.go(id=account.id)
                        account.opening_date = self.page.get_opening_date()
                        accounts.append(account)
                        page.go()

                    elif account.type == Account.TYPE_PERP:
                        # PERP balances must be fetched from the details page,
                        # otherwise we just scrape the "Rente annuelle estimée":
                        balance_page = self.open(account.url).page
                        # Sometimes the balance page is not available (with the site saying the account detail will
                        # soon be available). In the meantime, just skip it and use the balance we already have.
                        if balance_page and isinstance(balance_page, SavingAccountSummary):
                            balance = balance_page.get_balance()
                            if balance is not None:
                                account.balance = balance
                        else:
                            self.logger.debug('Balance page not yet available for account %r', account)
                    else:
                        accounts.append(account)
                        if account.type == Account.TYPE_CHECKING and account._has_cards:
                            to_check.append(account)
                if self.page.has_mandate_management_space:
                    self.location(self.page.mandate_management_space_link())
                    for mandate_account in self.page.iter_accounts():
                        accounts.append(mandate_account)

                for account in to_check:
                    accounts.extend(self.iter_cards(account))
                to_check = []

            if mandate_urls:
                for url in mandate_urls:
                    self.location(url)
                    if self.mandate_accounts_list.is_here():
                        for account in self.page.iter_accounts():
                            accounts.append(account)

            self.accounts = accounts
            # if we are sure there is no accounts on the all visited pages,
            # it is legit.
            if no_accounts == len(pages):
                raise NoAccountsException()
        return self.accounts
    @need_login
    def fill_account_ownership(self, account):
        profile = self.get_profile()
        if not account._account_holder or not profile:
        owner_name = profile.name
        pattern = re.compile(
            r'(m|mr|me|mme|mlle|mle|ml)\.? (.*)\bou ?(m|mr|me|mme|mlle|mle|ml)?\b(.*)',
            re.IGNORECASE
        )
        if pattern.search(account._account_holder):
            account.ownership = AccountOwnership.CO_OWNER
        elif all(
            n in account._account_holder
            for n in owner_name.lower().split(' ')
        ):
            account.ownership = AccountOwnership.OWNER
        else:
            account.ownership = AccountOwnership.ATTORNEY

    def get_loans(self, account):
        loans = []
        self.location(account.url)
        if 'initSSO' not in account.url:
            if self.temporary_page.is_here():
                # The main way to access to all "PRÊT PERSONNEL" seems to be broken
                # We must follow the following urls for cookies
                self.location(self.page.get_next_link())
                self.temporary_page.go()
                self.location(self.page.get_next_link())
                self.personal_loan_routage_url.go()
                self.page.form_submit()

                if self.par_accounts_loan.is_here():
                        self.logger.warning('Details are not available for this loans account: %s', account.id)
                    else:
                        loan = self.page.get_personal_loan()
                        if loan is not None:
                            # These Loans were not returned before
                            # So if they repair the precedent behaviour
                            # we must check where to get them
                            loan.id = loan.number = account.id
                            loan.label = account.label
                            loan.currency = account.currency
                            loan.url = account.url
                            loans.append(loan)
            else:
                for loan in self.page.iter_loans():
                    loan.currency = account.currency
                    loans.append(loan)
                student_loan = self.page.get_student_loan()
                if student_loan:
                    # Number of headers and item elements are the same
                    assert len(student_loan._heads) == len(student_loan._items)
                    student_loan.currency = account.currency
                    loans.append(student_loan)
        else:
            revolving = self.switch_account_to_revolving(account)

            self.revolving_redirection.go()
            self.page.fill_revolving(obj=revolving)

            loans.append(revolving)
    def switch_account_to_revolving(self, account):
        revolving = Loan()

        revolving.id = account.id
        revolving.label = f'{account.label} - {account.id}'
        revolving.currency = account.currency
        revolving.url = account.url
        revolving.owner_type = AccountOwnerType.PRIVATE
        revolving._has_cards = False
        revolving.type = Account.TYPE_REVOLVING_CREDIT

        return revolving
    def iter_cards(self, account):
        self.deferred_card_history.go(accountId=account.id, monthIndex=0, cardIndex=0)
        if self.cards_list.is_here():
            self.logger.debug('multiple cards for account %r', account)
            for card in self.page.get_cards(parent_id=account.id):
                card.parent = account
                yield card
        elif not account._has_deferred_history:
            # If the deferred card has no history available, we should use the API call to get details
            # otherwise trying to access the card history will make the website crash
            self.logger.debug('card with no history for account %r' % account)
            self.cards_json_detail.go(
                account_id=account.id,
                headers={'DISFE-CCX-Code-Appelant': '0004'}
            )
            for card in self.page.iter_cards(parent_id=account.id):
                card.parent = account
                yield card
            # The website does not shows the transactions if we do not
            # redirect to a precedent month with woob then come back
            self.single_card_history.go(monthIndex=1)
            self.single_card_history.go(monthIndex=0)
            self.logger.debug('single card for account %r', account)
            self.logger.debug('parsing %r', self.url)
            card = self.page.get_single_card(parent_id=account.id)
            card.parent = account
            yield card

    @need_login
    def get_history(self, account):
        if account.type == Account.TYPE_CHECKING and account.balance == 0:
            # When the balance is 0, we get a website unavailable on the history page
            # and the following navigation is broken
            return []

        if account.type == Account.TYPE_CARD and not account.url:
            # If the account is a deferred card with no available history, only return one transaction,
            # the card summary transaction
            self.logger.debug('creating card summary transaction for account %r' % account)
            self.cards_json_detail.go(
                account_id=account.parent.id,
                headers={'DISFE-CCX-Code-Appelant': '0004'}
            )
            return self.page.generate_summary(account)

        # TODO scrap pdf to get history of mandate accounts
        if account.url and 'gestion-sous-mandat' in account.url:
        if account.type in (account.TYPE_PEA, account.TYPE_MARKET):
            self.go_linebourse(account)
            return self.linebourse.iter_history(account.id)

Heddy's avatar
Heddy committed
        if account.type in (
            Account.TYPE_LOAN,
            Account.TYPE_REVOLVING_CREDIT,
            Account.TYPE_MORTGAGE,
        ):
        if account.type == Account.TYPE_CARD:
            return (tr for tr in self.iter_card_transactions(account) if not tr._coming)
        else:
            self.location(account.url)
ntome's avatar
ntome committed
            history_pages = {
                Account.TYPE_CHECKING: self.par_account_checking_history,
                Account.TYPE_SAVINGS: self.par_account_savings_and_invests_history,
                Account.TYPE_MARKET: self.par_account_savings_and_invests_history,
            }
            history = history_pages.get(account.type)
            if history is not None and account.label.casefold() != 'compte attente':
            # TODO be smarter by avoid fetching all, sorting all and returning all if only coming were desired
            if hasattr(self.page, 'iter_transactions') and self.page.has_transactions():
                return self.page.iter_transactions()
            elif account.type in (Account.TYPE_PERP, Account.TYPE_LIFE_INSURANCE) and self.retirement_hist.is_here():
            elif account.type == Account.TYPE_LIFE_INSURANCE:
                self.lifeinsurance_init_page.go()  # Mandatory for lifeinsurance_history_access_page
                self.lifeinsurance_access_page.go()
                product_code = self.page.get_product_code()
                assert product_code, 'product_code not found'
                self.page.submit_form()
                # Here we're redirected on a URL which has the contract_id as a param
                contract_id = get_url_param(self.url, 'identifiantContrat')
                assert contract_id, 'contract_id not found'
                params = {
                    'dateDebut': (datetime.now() - relativedelta(years=3)).strftime('%Y-%m-%d'),
                    'dateFin': datetime.now().strftime('%Y-%m-%d'),
                }
                    self.lifeinsurance_history.go(
                        contract_id=contract_id,
                        product_code=product_code,
                        params=params,
                    )
                    return self.page.get_history()
                except BrowserUnavailable:
                    # "Unavailable website" message
                    # This page is unavailable for this contract
                    pass
            return []
Fong NGO's avatar
Fong NGO committed
    def update_linebourse_session(self):
        self.linebourse.session.cookies.update(self.session.cookies)
        self.linebourse.session.headers['X-XSRF-TOKEN'] = self.session.cookies.get(
            'XSRF-TOKEN',
            domain='labanquepostale.offrebourse.com',
        )

    @need_login
    def go_linebourse(self, account):
        # Sometimes the redirection done from MarketLoginPage
        # (to https://labanquepostale.offrebourse.com/ReroutageSJR)
        # throws an error 404 or 403
        location = retry(HTTPError, delay=5)(self.location)
        location(account.url)

        # TODO Might be deprecated, check the logs after some time to
        # check if this is still used.
        if not self.market_home.is_here():
            self.logger.debug('Landed in unexpected market page, doing self.market_login.go()')
            go = retry(HTTPError, delay=5)(self.market_login.go)
            go()

Fong NGO's avatar
Fong NGO committed
        self.update_linebourse_session()
        self.par_accounts_checking.go()

    def _get_coming_transactions(self, account):
        if account.type == Account.TYPE_CHECKING:
            self.location(account.url)
            self.par_account_checking_coming.go(accountId=account.id)
            if self.par_account_checking_coming.is_here() and self.page.has_transactions():
                for tr in self.page.iter_transactions(coming=True):
                    yield tr
    @need_login
    def get_coming(self, account):
        if not account.url or 'gestion-sous-mandat' in account.url:
            return []
        # When the balance is 0, we get a website unavailable on the history page
        # and the following navigation is broken
        if account.type == Account.TYPE_CHECKING and account.balance != 0:
            return self._get_coming_transactions(account)
        elif account.type == Account.TYPE_CARD:
            transactions = []
            for tr in self.iter_card_transactions(account):
                if tr._coming:
                    transactions.append(tr)
            return transactions
        return []
    @need_login
    def iter_card_transactions(self, account):
        def iter_transactions(link, urlobj):
            # we go back to main menue otherwise we get an error 500.
            self.cards_list.go(account_id=account.parent.id)
            assert urlobj.is_here()
            ncard = self.page.params.get('cardIndex', 0)
            self.logger.debug('handling card %s for account %r', ncard, account)
            urlobj.go(accountId=account.parent.id, monthIndex=1, cardIndex=ncard)
                    urlobj.go(accountId=account.parent.id, monthIndex=t, cardIndex=ncard)
                except BrowserUnavailable:
                    self.logger.debug("deferred card history stop at %s", t)
                    break
                if urlobj.is_here():
                    for tr in self.page.get_history(deferred=True):
                        yield tr

        assert account.type == Account.TYPE_CARD
        for tr in iter_transactions(account.url, self.deferred_card_history_multi):
            yield tr
    @need_login
    def iter_investment(self, account):
        if account.url and 'gestion-sous-mandat' in account.url:
            return self.location(account.url).page.iter_investments()

        if account.type in (account.TYPE_PEA, account.TYPE_MARKET):
            self.go_linebourse(account)
            return self.linebourse.iter_investments(account.id)
ntome's avatar
ntome committed
        if account.type != Account.TYPE_LIFE_INSURANCE:
        self.location(account.url)

        self.lifeinsurance_init_page.go()  # Mandatory for lifeinsurance_history_access_page
        self.lifeinsurance_access_page.go()
        product_code = self.page.get_product_code()
        assert product_code, 'product_code not found'
        self.page.submit_form()
        # Here we're redirected on a URL which has the contract_id as a param
        contract_id = get_url_param(self.url, 'identifiantContrat')
        assert contract_id, 'contract_id not found'

        self.lifeinsurance_invest.go(
            contract_id=contract_id,
            product_code=product_code,
        )
        investments = list(self.page.iter_investments())
        return investments
    @need_login
    def iter_market_orders(self, account):
        if account.type not in (account.TYPE_PEA, account.TYPE_MARKET):
            return []

        self.go_linebourse(account)
        return self.linebourse.iter_market_orders(account.id)

    @need_login
    def iter_recipients(self, account_id):
        return self.transfer_choose.stay_or_go().iter_recipients(account_id=account_id)
    def init_transfer(self, account, recipient, amount, transfer):
        self.transfer_choose.stay_or_go()
Célande Adrien's avatar
Célande Adrien committed
        self.page.init_transfer(account.id, recipient._value, amount)

        assert self.transfer_complete.is_here(), 'An error occured while validating the first part of the transfer.'

        # Happens when making a transfer from a savings account
        # to an external account.
        # The transfer will be blocked with an error popup related to loi 6902
        blocage_popin_suffix = self.page.get_blocage_popin_url_suffix()
        if blocage_popin_suffix:
            self.popup_loi6902_transfer.go(popin_suffix=blocage_popin_suffix)

            msg = self.page.get_popup_message()
            raise TransferInvalidRecipient(message=msg)
Célande Adrien's avatar
Célande Adrien committed
        self.page.complete_transfer(transfer)

        return self.page.handle_response(transfer)

    def validate_transfer_code(self, transfer, code):