[fulbank.git] / fulbank / fsb.py

import json, http.cookiejar, binascii, time, datetime, pickle, urllib.error
from urllib import request, parse
from bs4 import BeautifulSoup as soup
from . import currency, auth, data
soupify = lambda cont: soup(cont, "html.parser")

apibase = "https://online.swedbank.se/TDE_DAP_Portal_REST_WEB/api/"
loginurl = "https://online.swedbank.se/app/privat/login"
serviceid = "B7dZHQcY78VRVz9l"

class fmterror(Exception):
    pass

class autherror(auth.autherror):
    pass

class jsonerror(Exception):
    def __init__(self, code, data, headers):
        self.code = code
        self.data = data
        self.headers = headers

    @classmethod
    def fromerr(cls, err):
        cs = err.headers.get_content_charset()
        if cs is None:
            cs = "utf-8"
        data = json.loads(err.read().decode(cs))
        return cls(err.code, data, err.headers)

def resolve(d, keys, default=fmterror):
    def err(key):
        if default is fmterror:
            raise fmterror(key)
        return default
    def rec(d, keys):
        if len(keys) == 0:
            return d
        if isinstance(d, dict):
            if keys[0] not in d:
                return err(keys[0])
            return rec(d[keys[0]], keys[1:])
        elif isinstance(d, list):
            if not 0 <= keys[0] < len(d):
                return err(keys[0])
            return rec(d[keys[0]], keys[1:])
        else:
            return err(keys[0])
    return rec(d, keys)

def linkurl(ln):
    if ln[0] != '/':
        raise fmterror("unexpected link url: " + ln)
    return parse.urljoin(apibase, ln[1:])

def getdsid():
    with request.urlopen(loginurl) as resp:
        if resp.code != 200:
            raise fmterror("Unexpected HTTP status code: " + str(resp.code))
        doc = soupify(resp.read())
    dsel = doc.find("div", id="cust-sess-id")
    if not dsel or not dsel.has_attr("value"):
        raise fmterror("DSID DIV not on login page")
    return dsel["value"]

def base64(data):
    return binascii.b2a_base64(data).decode("ascii").strip().rstrip("=")

class transaction(data.transaction):
    def __init__(self, account, data):
        self.account = account
        self._data = data

    _datefmt = "%Y-%m-%d"

    @property
    def value(self): return currency.currency.get(resolve(self._data, ("currency",))).parse(resolve(self._data, ("amount",)))
    @property
    def message(self): return resolve(self._data, ("description",))
    @property
    def date(self):
        p = time.strptime(resolve(self._data, ("accountingDate",)), self._datefmt)
        return datetime.date(p.tm_year, p.tm_mon, p.tm_mday)

class txnaccount(data.txnaccount):
    def __init__(self, sess, id, idata):
        self.sess = sess
        self.id = id
        self._data = None
        self._idata = idata

    @property
    def data(self):
        if self._data is None:
            self._data = self.sess._jreq("v5/engagement/account/" + self.id)
        return self._data

    @property
    def number(self): return resolve(self.data, ("accountNumber",))
    @property
    def clearing(self): return resolve(self.data, ("clearingNumber",))
    @property
    def fullnumber(self): return resolve(self.data, ("fullyFormattedNumber",))
    @property
    def balance(self): return currency.currency.get(resolve(self.data, ("balance", "currencyCode"))).parse(resolve(self.data, ("balance", "amount")))
    @property
    def name(self): return resolve(self._idata, ("name",))

    def transactions(self):
        pagesz = 50
        page = 1
        while True:
            data = self.sess._jreq("v5/engagement/transactions/" + self.id, transactionsPerPage=pagesz, page=page)
            txlist = resolve(data, ("transactions",))
            if len(txlist) < 1:
                break
            for tx in txlist:
                yield transaction(self, tx)
            page += 1

class cardtransaction(data.transaction):
    def __init__(self, account, data):
        self.account = account
        self._data = data

    _datefmt = "%Y-%m-%d"

    @property
    def value(self):
        am = resolve(self._data, ("localAmount",))
        return currency.currency.get(resolve(am, ("currencyCode",))).parse(resolve(am, ("amount",)))
    @property
    def message(self): return resolve(self._data, ("description",))
    @property
    def date(self):
        p = time.strptime(resolve(self._data, ("date",)), self._datefmt)
        return datetime.date(p.tm_year, p.tm_mon, p.tm_mday)

class cardaccount(data.cardaccount):
    def __init__(self, sess, id, idata):
        self.sess = sess
        self.id = id
        self._data = None
        self._idata = idata

    @property
    def data(self):
        if self._data is None:
            self._data = self.sess._jreq("v5/engagement/cardaccount/" + self.id)
        return self._data

    @property
    def number(self): return resolve(self.data, ("cardAccount", "cardNumber"))
    @property
    def balance(self):
        cc = resolve(self.data, ("transactions", 0, "localAmount", "currencyCode"))
        return currency.currency.get(cc).parse(resolve(self.data, ("cardAccount", "currentBalance")))
    @property
    def name(self): return resolve(self._idata, ("name",))

    def transactions(self):
        pagesz = 50
        page = 1
        while True:
            data = self.sess._jreq("v5/engagement/cardaccount/" + self.id, transactionsPerPage=pagesz, page=page)
            txlist = resolve(data, ("transactions",))
            if len(txlist) < 1:
                break
            for tx in txlist:
                yield cardtransaction(self, tx)
            page += 1

class session(object):
    def __init__(self, dsid):
        self.dsid = dsid
        self.auth = base64((serviceid + ":" + str(int(time.time() * 1000))).encode("ascii"))
        self.jar = request.HTTPCookieProcessor()
        self.jar.cookiejar.set_cookie(http.cookiejar.Cookie(
            version=0, name="dsid", value=dsid, path="/", path_specified=True,
            domain=".online.swedbank.se", domain_specified=True, domain_initial_dot=True,
            port=None, port_specified=False, secure=False, expires=None,
            discard=True, comment=None, comment_url=None,
            rest={}, rfc2109=False))
        self.userid = None
        self._accounts = None

    def _req(self, url, data=None, ctype=None, headers={}, method=None, **kws):
        if "dsid" not in kws:
            kws["dsid"] = self.dsid
        kws = {k: v for (k, v) in kws.items() if v is not None}
        url = parse.urljoin(apibase, url + "?" + parse.urlencode(kws))
        if isinstance(data, dict):
            data = json.dumps(data).encode("utf-8")
            ctype = "application/json;charset=UTF-8"
        req = request.Request(url, data=data, method=method)
        for hnam, hval in headers.items():
            req.add_header(hnam, hval)
        if ctype is not None:
            req.add_header("Content-Type", ctype)
        req.add_header("Authorization", self.auth)
        self.jar.https_request(req)
        with request.urlopen(req) as resp:
            if resp.code != 200 and resp.code != 201:
                raise fmterror("Unexpected HTTP status code: " + str(resp.code))
            self.jar.https_response(req, resp)
            return resp.read()

    def _jreq(self, *args, **kwargs):
        headers = kwargs.pop("headers", {})
        headers["Accept"] = "application/json"
        try:
            ret = self._req(*args, headers=headers, **kwargs)
        except urllib.error.HTTPError as e:
            if e.headers.get_content_type() == "application/json":
                raise jsonerror.fromerr(e)
        return json.loads(ret.decode("utf-8"))

    def _postlogin(self):
        auth = self._jreq("v5/user/authenticationinfo")
        uid = auth.get("identifiedUser", "")
        if uid == "":
            raise fmterror("no identified user even after successful authentication")
        self.userid = uid
        prof = self._jreq("v5/profile/")
        if len(prof["banks"]) != 1:
            raise fmterror("do not know the meaning of multiple banks")
        rolesw = linkurl(resolve(prof["banks"][0], ("privateProfile", "links", "next", "uri")))
        self._jreq(rolesw, method="POST")

    def auth_bankid(self, user, conv=None):
        if conv is None:
            conv = auth.default()
        try:
            data = self._jreq("v5/identification/bankid/mobile", data = {
                "userId": user,
                "useEasyLogin": False,
                "generateEasyLoginId": False})
        except jsonerror as e:
            if e.code == 400:
                flds = resolve(e.data, ("errorMessages", "fields"), False)
                if isinstance(flds, list):
                    for fld in flds:
                        if resolve(fld, ("field",), None) == "userId":
                            raise autherror(fld["message"])
            raise
        st = data.get("status")
        vfy = linkurl(resolve(data, ("links", "next", "uri")))
        fst = None
        while True:
            if st in {"USER_SIGN", "CLIENT_NOT_STARTED"}:
                if st != fst:
                    conv.message("Status: %s" % (st,), auth.conv.msg_info)
                    fst = st
                continue
            elif st == "COMPLETE":
                self._postlogin()
                return
            elif st == "CANCELLED":
                raise autherror("authentication cancelled")
            elif st == "OUTSTANDING_TRANSACTION":
                raise autherror("another bankid transaction already in progress")
            else:
                raise fmterror("unexpected bankid status: " + str(st))
            time.sleep(3)
            vdat = self._jreq(vfy)
            st = vdat.get("status")

    def keepalive(self):
        data = self._jreq("v5/framework/clientsession")
        return data["timeoutInMillis"] / 1000

    @property
    def accounts(self):
        if self._accounts is None:
            data = self._jreq("v5/engagement/overview")
            accounts = []
            for acct in resolve(data, ("transactionAccounts",)):
                accounts.append(txnaccount(self, resolve(acct, ("id",)), acct))
            for acct in resolve(data, ("cardAccounts",)):
                accounts.append(cardaccount(self, resolve(acct, ("id",)), acct))
            self._accounts = accounts
        return self._accounts

    def logout(self):
        if self.userid is not None:
            self._jreq("v5/identification/logout", method="PUT")
            self.userid = None

    def close(self):
        self.logout()
        self._req("v5/framework/clientsession", method="DELETE")

    def __enter__(self):
        return self

    def __exit__(self, *excinfo):
        self.close()
        return False

    def __repr__(self):
        if self.userid is not None:
            return "#<fsb.session %s>" % self.userid
        return "#<fsb.session>"

    @classmethod
    def create(cls):
        return cls(getdsid())

    def save(self, filename):
        with open(filename, "wb") as fp:
            pickle.dump(self, fp)

    @classmethod
    def load(cls, filename):
        with open(filename, "rb") as fp:
            return pickle.load(fp)
Commit	Line	Data
f78b790d	1	import json, http.cookiejar, binascii, time, datetime, pickle, urllib.error
8e60b2da FT	2	from urllib import request, parse
8e60b2da FT	3	from bs4 import BeautifulSoup as soup
717d54fc	4	from . import currency, auth, data
8e60b2da FT	5	soupify = lambda cont: soup(cont, "html.parser")
	6
	7	apibase = "https://online.swedbank.se/TDE_DAP_Portal_REST_WEB/api/"
	8	loginurl = "https://online.swedbank.se/app/privat/login"
	9	serviceid = "B7dZHQcY78VRVz9l"
	10
	11	class fmterror(Exception):
	12	pass
	13
f99c3f74	14	class autherror(auth.autherror):
8e60b2da FT	15	pass
8e60b2da FT	16
f78b790d FT	17	class jsonerror(Exception):
	18	def __init__(self, code, data, headers):
	19	self.code = code
	20	self.data = data
	21	self.headers = headers
	22
	23	@classmethod
	24	def fromerr(cls, err):
	25	cs = err.headers.get_content_charset()
	26	if cs is None:
	27	cs = "utf-8"
	28	data = json.loads(err.read().decode(cs))
	29	return cls(err.code, data, err.headers)
	30
8e60b2da	31	def resolve(d, keys, default=fmterror):
d791f2f2	32	def err(key):
8e60b2da	33	if default is fmterror:
d791f2f2	34	raise fmterror(key)
8e60b2da FT	35	return default
	36	def rec(d, keys):
	37	if len(keys) == 0:
	38	return d
	39	if isinstance(d, dict):
	40	if keys[0] not in d:
d791f2f2 FT	41	return err(keys[0])
	42	return rec(d[keys[0]], keys[1:])
	43	elif isinstance(d, list):
	44	if not 0 <= keys[0] < len(d):
	45	return err(keys[0])
8e60b2da FT	46	return rec(d[keys[0]], keys[1:])
8e60b2da FT	47	else:
d791f2f2	48	return err(keys[0])
8e60b2da FT	49	return rec(d, keys)
	50
	51	def linkurl(ln):
	52	if ln[0] != '/':
	53	raise fmterror("unexpected link url: " + ln)
	54	return parse.urljoin(apibase, ln[1:])
	55
	56	def getdsid():
	57	with request.urlopen(loginurl) as resp:
	58	if resp.code != 200:
	59	raise fmterror("Unexpected HTTP status code: " + str(resp.code))
	60	doc = soupify(resp.read())
	61	dsel = doc.find("div", id="cust-sess-id")
	62	if not dsel or not dsel.has_attr("value"):
	63	raise fmterror("DSID DIV not on login page")
	64	return dsel["value"]
	65
	66	def base64(data):
	67	return binascii.b2a_base64(data).decode("ascii").strip().rstrip("=")
	68
717d54fc	69	class transaction(data.transaction):
61fd054f FT	70	def __init__(self, account, data):
	71	self.account = account
	72	self._data = data
	73
8c456209 FT	74	_datefmt = "%Y-%m-%d"
8c456209 FT	75
61fd054f	76	@property
8e415ee7	77	def value(self): return currency.currency.get(resolve(self._data, ("currency",))).parse(resolve(self._data, ("amount",)))
61fd054f	78	@property
8c456209 FT	79	def message(self): return resolve(self._data, ("description",))
	80	@property
	81	def date(self):
	82	p = time.strptime(resolve(self._data, ("accountingDate",)), self._datefmt)
	83	return datetime.date(p.tm_year, p.tm_mon, p.tm_mday)
	84
717d54fc	85	class txnaccount(data.txnaccount):
61fd054f FT	86	def __init__(self, sess, id, idata):
	87	self.sess = sess
	88	self.id = id
	89	self._data = None
	90	self._idata = idata
	91
	92	@property
	93	def data(self):
	94	if self._data is None:
	95	self._data = self.sess._jreq("v5/engagement/account/" + self.id)
	96	return self._data
	97
	98	@property
	99	def number(self): return resolve(self.data, ("accountNumber",))
	100	@property
	101	def clearing(self): return resolve(self.data, ("clearingNumber",))
	102	@property
	103	def fullnumber(self): return resolve(self.data, ("fullyFormattedNumber",))
	104	@property
0ba154d0 FT	105	def balance(self): return currency.currency.get(resolve(self.data, ("balance", "currencyCode"))).parse(resolve(self.data, ("balance", "amount")))
0ba154d0 FT	106	@property
61fd054f FT	107	def name(self): return resolve(self._idata, ("name",))
	108
	109	def transactions(self):
	110	pagesz = 50
8c456209 FT	111	page = 1
	112	while True:
	113	data = self.sess._jreq("v5/engagement/transactions/" + self.id, transactionsPerPage=pagesz, page=page)
	114	txlist = resolve(data, ("transactions",))
	115	if len(txlist) < 1:
	116	break
	117	for tx in txlist:
	118	yield transaction(self, tx)
	119	page += 1
61fd054f	120
717d54fc	121	class cardtransaction(data.transaction):
d791f2f2 FT	122	def __init__(self, account, data):
	123	self.account = account
	124	self._data = data
	125
	126	_datefmt = "%Y-%m-%d"
	127
	128	@property
	129	def value(self):
	130	am = resolve(self._data, ("localAmount",))
	131	return currency.currency.get(resolve(am, ("currencyCode",))).parse(resolve(am, ("amount",)))
	132	@property
	133	def message(self): return resolve(self._data, ("description",))
	134	@property
	135	def date(self):
	136	p = time.strptime(resolve(self._data, ("date",)), self._datefmt)
	137	return datetime.date(p.tm_year, p.tm_mon, p.tm_mday)
	138
717d54fc	139	class cardaccount(data.cardaccount):
d791f2f2 FT	140	def __init__(self, sess, id, idata):
	141	self.sess = sess
	142	self.id = id
	143	self._data = None
	144	self._idata = idata
	145
	146	@property
	147	def data(self):
	148	if self._data is None:
	149	self._data = self.sess._jreq("v5/engagement/cardaccount/" + self.id)
	150	return self._data
	151
	152	@property
	153	def number(self): return resolve(self.data, ("cardAccount", "cardNumber"))
	154	@property
	155	def balance(self):
	156	cc = resolve(self.data, ("transactions", 0, "localAmount", "currencyCode"))
	157	return currency.currency.get(cc).parse(resolve(self.data, ("cardAccount", "currentBalance")))
	158	@property
	159	def name(self): return resolve(self._idata, ("name",))
	160
	161	def transactions(self):
	162	pagesz = 50
	163	page = 1
	164	while True:
	165	data = self.sess._jreq("v5/engagement/cardaccount/" + self.id, transactionsPerPage=pagesz, page=page)
	166	txlist = resolve(data, ("transactions",))
	167	if len(txlist) < 1:
	168	break
	169	for tx in txlist:
	170	yield cardtransaction(self, tx)
	171	page += 1
	172
8e60b2da FT	173	class session(object):
	174	def __init__(self, dsid):
	175	self.dsid = dsid
	176	self.auth = base64((serviceid + ":" + str(int(time.time() * 1000))).encode("ascii"))
	177	self.jar = request.HTTPCookieProcessor()
	178	self.jar.cookiejar.set_cookie(http.cookiejar.Cookie(
	179	version=0, name="dsid", value=dsid, path="/", path_specified=True,
	180	domain=".online.swedbank.se", domain_specified=True, domain_initial_dot=True,
	181	port=None, port_specified=False, secure=False, expires=None,
	182	discard=True, comment=None, comment_url=None,
	183	rest={}, rfc2109=False))
	184	self.userid = None
61fd054f	185	self._accounts = None
8e60b2da FT	186
	187	def _req(self, url, data=None, ctype=None, headers={}, method=None, **kws):
	188	if "dsid" not in kws:
	189	kws["dsid"] = self.dsid
	190	kws = {k: v for (k, v) in kws.items() if v is not None}
	191	url = parse.urljoin(apibase, url + "?" + parse.urlencode(kws))
	192	if isinstance(data, dict):
	193	data = json.dumps(data).encode("utf-8")
	194	ctype = "application/json;charset=UTF-8"
	195	req = request.Request(url, data=data, method=method)
	196	for hnam, hval in headers.items():
	197	req.add_header(hnam, hval)
	198	if ctype is not None:
	199	req.add_header("Content-Type", ctype)
	200	req.add_header("Authorization", self.auth)
	201	self.jar.https_request(req)
	202	with request.urlopen(req) as resp:
61fd054f	203	if resp.code != 200 and resp.code != 201:
8e60b2da FT	204	raise fmterror("Unexpected HTTP status code: " + str(resp.code))
	205	self.jar.https_response(req, resp)
	206	return resp.read()
	207
	208	def _jreq(self, args, *kwargs):
	209	headers = kwargs.pop("headers", {})
	210	headers["Accept"] = "application/json"
f78b790d FT	211	try:
	212	ret = self._req(args, headers=headers, *kwargs)
	213	except urllib.error.HTTPError as e:
	214	if e.headers.get_content_type() == "application/json":
	215	raise jsonerror.fromerr(e)
8e60b2da FT	216	return json.loads(ret.decode("utf-8"))
8e60b2da FT	217
61fd054f FT	218	def _postlogin(self):
	219	auth = self._jreq("v5/user/authenticationinfo")
	220	uid = auth.get("identifiedUser", "")
	221	if uid == "":
	222	raise fmterror("no identified user even after successful authentication")
	223	self.userid = uid
	224	prof = self._jreq("v5/profile/")
	225	if len(prof["banks"]) != 1:
	226	raise fmterror("do not know the meaning of multiple banks")
	227	rolesw = linkurl(resolve(prof["banks"][0], ("privateProfile", "links", "next", "uri")))
	228	self._jreq(rolesw, method="POST")
	229
f4de0bf1 FT	230	def auth_bankid(self, user, conv=None):
	231	if conv is None:
	232	conv = auth.default()
f78b790d FT	233	try:
	234	data = self._jreq("v5/identification/bankid/mobile", data = {
	235	"userId": user,
	236	"useEasyLogin": False,
	237	"generateEasyLoginId": False})
	238	except jsonerror as e:
	239	if e.code == 400:
	240	flds = resolve(e.data, ("errorMessages", "fields"), False)
	241	if isinstance(flds, list):
	242	for fld in flds:
	243	if resolve(fld, ("field",), None) == "userId":
	244	raise autherror(fld["message"])
	245	raise
8cda37c5	246	st = data.get("status")
8e60b2da	247	vfy = linkurl(resolve(data, ("links", "next", "uri")))
f4de0bf1	248	fst = None
8e60b2da	249	while True:
f4de0bf1 FT	250	if st in {"USER_SIGN", "CLIENT_NOT_STARTED"}:
	251	if st != fst:
	252	conv.message("Status: %s" % (st,), auth.conv.msg_info)
	253	fst = st
61fd054f	254	continue
8e60b2da	255	elif st == "COMPLETE":
61fd054f	256	self._postlogin()
8e60b2da FT	257	return
	258	elif st == "CANCELLED":
	259	raise autherror("authentication cancelled")
86601bc0 FT	260	elif st == "OUTSTANDING_TRANSACTION":
86601bc0 FT	261	raise autherror("another bankid transaction already in progress")
8e60b2da FT	262	else:
8e60b2da FT	263	raise fmterror("unexpected bankid status: " + str(st))
8cda37c5 FT	264	time.sleep(3)
	265	vdat = self._jreq(vfy)
	266	st = vdat.get("status")
8e60b2da	267
61fd054f FT	268	def keepalive(self):
	269	data = self._jreq("v5/framework/clientsession")
	270	return data["timeoutInMillis"] / 1000
	271
	272	@property
	273	def accounts(self):
	274	if self._accounts is None:
	275	data = self._jreq("v5/engagement/overview")
	276	accounts = []
	277	for acct in resolve(data, ("transactionAccounts",)):
d791f2f2 FT	278	accounts.append(txnaccount(self, resolve(acct, ("id",)), acct))
	279	for acct in resolve(data, ("cardAccounts",)):
	280	accounts.append(cardaccount(self, resolve(acct, ("id",)), acct))
61fd054f FT	281	self._accounts = accounts
	282	return self._accounts
	283
8e60b2da FT	284	def logout(self):
	285	if self.userid is not None:
	286	self._jreq("v5/identification/logout", method="PUT")
	287	self.userid = None
	288
	289	def close(self):
	290	self.logout()
	291	self._req("v5/framework/clientsession", method="DELETE")
	292
	293	def __enter__(self):
	294	return self
	295
	296	def __exit__(self, *excinfo):
	297	self.close()
	298	return False
	299
61fd054f FT	300	def __repr__(self):
	301	if self.userid is not None:
	302	return "#<fsb.session %s>" % self.userid
	303	return "#<fsb.session>"
	304
8e60b2da FT	305	@classmethod
	306	def create(cls):
	307	return cls(getdsid())
61fd054f FT	308
	309	def save(self, filename):
	310	with open(filename, "wb") as fp:
	311	pickle.dump(self, fp)
	312
	313	@classmethod
	314	def load(cls, filename):
	315	with open(filename, "rb") as fp:
c526374d	316	return pickle.load(fp)