1 import json, http.cookiejar, binascii, time, pickle
2 from urllib import request, parse
3 from bs4 import BeautifulSoup as soup
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:])
35 raise fmterror("unexpected link url: " + ln)
36 return parse.urljoin(apibase, ln[1:])
39 with request.urlopen(loginurl) as resp:
41 raise fmterror("Unexpected HTTP status code: " + str(resp.code))
42 doc = soupify(resp.read())
43 dsel = doc.find("div", id="cust-sess-id")
44 if not dsel or not dsel.has_attr("value"):
45 raise fmterror("DSID DIV not on login page")
49 return binascii.b2a_base64(data).decode("ascii").strip().rstrip("=")
51 class transaction(object):
52 def __init__(self, account, data):
53 self.account = account
57 def value(self): return currency.currency.get(resolve(self._data, ("currency",))).parse(resolve(self._data, ("amount",)))
59 def message(self): return resolve(self._data, ("details", "message"))
62 return "#<fsb.transaction %s: %r>" % (self.value, self.message)
64 class account(object):
65 def __init__(self, sess, id, idata):
73 if self._data is None:
74 self._data = self.sess._jreq("v5/engagement/account/" + self.id)
78 def number(self): return resolve(self.data, ("accountNumber",))
80 def clearing(self): return resolve(self.data, ("clearingNumber",))
82 def fullnumber(self): return resolve(self.data, ("fullyFormattedNumber",))
84 def name(self): return resolve(self._idata, ("name",))
86 def transactions(self):
88 data = self.sess._jreq("v5/engagement/transactions/" + self.id, transactionsPerPage=pagesz, page=1)
89 for tx in resolve(data, ("transactions",)):
90 yield transaction(self, tx)
93 return "#<fsb.account %s: %r>" % (self.fullnumber, self.name)
95 class session(object):
96 def __init__(self, dsid):
98 self.auth = base64((serviceid + ":" + str(int(time.time() * 1000))).encode("ascii"))
99 self.jar = request.HTTPCookieProcessor()
100 self.jar.cookiejar.set_cookie(http.cookiejar.Cookie(
101 version=0, name="dsid", value=dsid, path="/", path_specified=True,
102 domain=".online.swedbank.se", domain_specified=True, domain_initial_dot=True,
103 port=None, port_specified=False, secure=False, expires=None,
104 discard=True, comment=None, comment_url=None,
105 rest={}, rfc2109=False))
107 self._accounts = None
109 def _req(self, url, data=None, ctype=None, headers={}, method=None, **kws):
110 if "dsid" not in kws:
111 kws["dsid"] = self.dsid
112 kws = {k: v for (k, v) in kws.items() if v is not None}
113 url = parse.urljoin(apibase, url + "?" + parse.urlencode(kws))
114 if isinstance(data, dict):
115 data = json.dumps(data).encode("utf-8")
116 ctype = "application/json;charset=UTF-8"
117 req = request.Request(url, data=data, method=method)
118 for hnam, hval in headers.items():
119 req.add_header(hnam, hval)
120 if ctype is not None:
121 req.add_header("Content-Type", ctype)
122 req.add_header("Authorization", self.auth)
123 self.jar.https_request(req)
124 with request.urlopen(req) as resp:
125 if resp.code != 200 and resp.code != 201:
126 raise fmterror("Unexpected HTTP status code: " + str(resp.code))
127 self.jar.https_response(req, resp)
130 def _jreq(self, *args, **kwargs):
131 headers = kwargs.pop("headers", {})
132 headers["Accept"] = "application/json"
133 ret = self._req(*args, headers=headers, **kwargs)
134 return json.loads(ret.decode("utf-8"))
136 def _postlogin(self):
137 auth = self._jreq("v5/user/authenticationinfo")
138 uid = auth.get("identifiedUser", "")
140 raise fmterror("no identified user even after successful authentication")
142 prof = self._jreq("v5/profile/")
143 if len(prof["banks"]) != 1:
144 raise fmterror("do not know the meaning of multiple banks")
145 rolesw = linkurl(resolve(prof["banks"][0], ("privateProfile", "links", "next", "uri")))
146 self._jreq(rolesw, method="POST")
148 def auth_bankid(self, user):
149 data = self._jreq("v5/identification/bankid/mobile", data = {
151 "useEasyLogin": False,
152 "generateEasyLoginId": False})
153 if data.get("status") != "USER_SIGN":
154 raise fmterror("unexpected bankid status: " + str(data.get("status")))
155 vfy = linkurl(resolve(data, ("links", "next", "uri")))
158 vdat = self._jreq(vfy)
159 st = vdat.get("status")
160 if st == "USER_SIGN":
162 elif st == "CLIENT_NOT_STARTED":
164 elif st == "COMPLETE":
167 elif st == "CANCELLED":
168 raise autherror("authentication cancelled")
170 raise fmterror("unexpected bankid status: " + str(st))
173 data = self._jreq("v5/framework/clientsession")
174 return data["timeoutInMillis"] / 1000
178 if self._accounts is None:
179 data = self._jreq("v5/engagement/overview")
181 for acct in resolve(data, ("transactionAccounts",)):
182 accounts.append(account(self, resolve(acct, ("id",)), acct))
183 self._accounts = accounts
184 return self._accounts
187 if self.userid is not None:
188 self._jreq("v5/identification/logout", method="PUT")
193 self._req("v5/framework/clientsession", method="DELETE")
198 def __exit__(self, *excinfo):
203 if self.userid is not None:
204 return "#<fsb.session %s>" % self.userid
205 return "#<fsb.session>"
209 return cls(getdsid())
211 def save(self, filename):
212 with open(filename, "wb") as fp:
213 pickle.dump(self, fp)
216 def load(cls, filename):
217 with open(filename, "rb") as fp:
218 return picke.load(fp)