1 import json, http.cookiejar, binascii, time, datetime, pickle, hashlib
2 from urllib import request, parse
3 from bs4 import BeautifulSoup as soup
4 from . import currency, auth
5 soupify = lambda cont: soup(cont, "html.parser")
7 apibase = "https://online.swedbank.se/TDE_DAP_Portal_REST_WEB/api/"
8 loginurl = "https://online.swedbank.se/app/privat/login"
9 serviceid = "B7dZHQcY78VRVz9l"
11 class fmterror(Exception):
14 class autherror(Exception):
17 def resolve(d, keys, default=fmterror):
19 if default is fmterror:
25 if isinstance(d, dict):
28 return rec(d[keys[0]], keys[1:])
29 elif isinstance(d, list):
30 if not 0 <= keys[0] < len(d):
32 return rec(d[keys[0]], keys[1:])
39 raise fmterror("unexpected link url: " + ln)
40 return parse.urljoin(apibase, ln[1:])
43 with request.urlopen(loginurl) as resp:
45 raise fmterror("Unexpected HTTP status code: " + str(resp.code))
46 doc = soupify(resp.read())
47 dsel = doc.find("div", id="cust-sess-id")
48 if not dsel or not dsel.has_attr("value"):
49 raise fmterror("DSID DIV not on login page")
53 return binascii.b2a_base64(data).decode("ascii").strip().rstrip("=")
55 class transaction(object):
56 def __init__(self, account, data):
57 self.account = account
63 def value(self): return currency.currency.get(resolve(self._data, ("currency",))).parse(resolve(self._data, ("amount",)))
65 def message(self): return resolve(self._data, ("description",))
68 p = time.strptime(resolve(self._data, ("accountingDate",)), self._datefmt)
69 return datetime.date(p.tm_year, p.tm_mon, p.tm_mday)
73 dig = hashlib.sha256()
74 dig.update(str(self.date.toordinal()).encode("ascii") + b"\0")
75 dig.update(self.message.encode("utf-8") + b"\0")
76 dig.update(str(self.value.amount).encode("ascii") + b"\0")
77 dig.update(self.value.currency.symbol.encode("ascii") + b"\0")
78 return dig.hexdigest()
81 return "#<fsb.transaction %s: %r>" % (self.value, self.message)
83 class txnaccount(object):
84 def __init__(self, sess, id, idata):
92 if self._data is None:
93 self._data = self.sess._jreq("v5/engagement/account/" + self.id)
97 def number(self): return resolve(self.data, ("accountNumber",))
99 def clearing(self): return resolve(self.data, ("clearingNumber",))
101 def fullnumber(self): return resolve(self.data, ("fullyFormattedNumber",))
103 def balance(self): return currency.currency.get(resolve(self.data, ("balance", "currencyCode"))).parse(resolve(self.data, ("balance", "amount")))
105 def name(self): return resolve(self._idata, ("name",))
107 def transactions(self):
111 data = self.sess._jreq("v5/engagement/transactions/" + self.id, transactionsPerPage=pagesz, page=page)
112 txlist = resolve(data, ("transactions",))
116 yield transaction(self, tx)
120 return "#<fsb.txnaccount %s: %r>" % (self.fullnumber, self.name)
122 class cardtransaction(object):
123 def __init__(self, account, data):
124 self.account = account
127 _datefmt = "%Y-%m-%d"
131 am = resolve(self._data, ("localAmount",))
132 return currency.currency.get(resolve(am, ("currencyCode",))).parse(resolve(am, ("amount",)))
134 def message(self): return resolve(self._data, ("description",))
137 p = time.strptime(resolve(self._data, ("date",)), self._datefmt)
138 return datetime.date(p.tm_year, p.tm_mon, p.tm_mday)
142 dig = hashlib.sha256()
143 dig.update(str(self.date.toordinal()).encode("ascii") + b"\0")
144 dig.update(self.message.encode("utf-8") + b"\0")
145 dig.update(str(self.value.amount).encode("ascii") + b"\0")
146 dig.update(self.value.currency.symbol.encode("ascii") + b"\0")
147 return dig.hexdigest()
150 return "#<fsb.cardtransaction %s: %r>" % (self.value, self.message)
152 class cardaccount(object):
153 def __init__(self, sess, id, idata):
161 if self._data is None:
162 self._data = self.sess._jreq("v5/engagement/cardaccount/" + self.id)
166 def number(self): return resolve(self.data, ("cardAccount", "cardNumber"))
169 cc = resolve(self.data, ("transactions", 0, "localAmount", "currencyCode"))
170 return currency.currency.get(cc).parse(resolve(self.data, ("cardAccount", "currentBalance")))
172 def name(self): return resolve(self._idata, ("name",))
174 def transactions(self):
178 data = self.sess._jreq("v5/engagement/cardaccount/" + self.id, transactionsPerPage=pagesz, page=page)
179 txlist = resolve(data, ("transactions",))
183 yield cardtransaction(self, tx)
187 return "#<fsb.cardaccount %s: %r>" % (self.number, self.name)
189 class session(object):
190 def __init__(self, dsid):
192 self.auth = base64((serviceid + ":" + str(int(time.time() * 1000))).encode("ascii"))
193 self.jar = request.HTTPCookieProcessor()
194 self.jar.cookiejar.set_cookie(http.cookiejar.Cookie(
195 version=0, name="dsid", value=dsid, path="/", path_specified=True,
196 domain=".online.swedbank.se", domain_specified=True, domain_initial_dot=True,
197 port=None, port_specified=False, secure=False, expires=None,
198 discard=True, comment=None, comment_url=None,
199 rest={}, rfc2109=False))
201 self._accounts = None
203 def _req(self, url, data=None, ctype=None, headers={}, method=None, **kws):
204 if "dsid" not in kws:
205 kws["dsid"] = self.dsid
206 kws = {k: v for (k, v) in kws.items() if v is not None}
207 url = parse.urljoin(apibase, url + "?" + parse.urlencode(kws))
208 if isinstance(data, dict):
209 data = json.dumps(data).encode("utf-8")
210 ctype = "application/json;charset=UTF-8"
211 req = request.Request(url, data=data, method=method)
212 for hnam, hval in headers.items():
213 req.add_header(hnam, hval)
214 if ctype is not None:
215 req.add_header("Content-Type", ctype)
216 req.add_header("Authorization", self.auth)
217 self.jar.https_request(req)
218 with request.urlopen(req) as resp:
219 if resp.code != 200 and resp.code != 201:
220 raise fmterror("Unexpected HTTP status code: " + str(resp.code))
221 self.jar.https_response(req, resp)
224 def _jreq(self, *args, **kwargs):
225 headers = kwargs.pop("headers", {})
226 headers["Accept"] = "application/json"
227 ret = self._req(*args, headers=headers, **kwargs)
228 return json.loads(ret.decode("utf-8"))
230 def _postlogin(self):
231 auth = self._jreq("v5/user/authenticationinfo")
232 uid = auth.get("identifiedUser", "")
234 raise fmterror("no identified user even after successful authentication")
236 prof = self._jreq("v5/profile/")
237 if len(prof["banks"]) != 1:
238 raise fmterror("do not know the meaning of multiple banks")
239 rolesw = linkurl(resolve(prof["banks"][0], ("privateProfile", "links", "next", "uri")))
240 self._jreq(rolesw, method="POST")
242 def auth_bankid(self, user, conv=None):
244 conv = auth.default()
245 data = self._jreq("v5/identification/bankid/mobile", data = {
247 "useEasyLogin": False,
248 "generateEasyLoginId": False})
249 if data.get("status") != "USER_SIGN":
250 raise fmterror("unexpected bankid status: " + str(data.get("status")))
251 vfy = linkurl(resolve(data, ("links", "next", "uri")))
255 vdat = self._jreq(vfy)
256 st = vdat.get("status")
257 if st in {"USER_SIGN", "CLIENT_NOT_STARTED"}:
259 conv.message("Status: %s" % (st,), auth.conv.msg_info)
262 elif st == "COMPLETE":
265 elif st == "CANCELLED":
266 raise autherror("authentication cancelled")
268 raise fmterror("unexpected bankid status: " + str(st))
271 data = self._jreq("v5/framework/clientsession")
272 return data["timeoutInMillis"] / 1000
276 if self._accounts is None:
277 data = self._jreq("v5/engagement/overview")
279 for acct in resolve(data, ("transactionAccounts",)):
280 accounts.append(txnaccount(self, resolve(acct, ("id",)), acct))
281 for acct in resolve(data, ("cardAccounts",)):
282 accounts.append(cardaccount(self, resolve(acct, ("id",)), acct))
283 self._accounts = accounts
284 return self._accounts
287 if self.userid is not None:
288 self._jreq("v5/identification/logout", method="PUT")
293 self._req("v5/framework/clientsession", method="DELETE")
298 def __exit__(self, *excinfo):
303 if self.userid is not None:
304 return "#<fsb.session %s>" % self.userid
305 return "#<fsb.session>"
309 return cls(getdsid())
311 def save(self, filename):
312 with open(filename, "wb") as fp:
313 pickle.dump(self, fp)
316 def load(cls, filename):
317 with open(filename, "rb") as fp:
318 return pickle.load(fp)