fulbank/fsb.py

   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")
   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
  14 class autherror(Exception):
  15     pass
  16
  17 def resolve(d, keys, default=fmterror):
  18     def err(key):
  19         if default is fmterror:
  20             raise fmterror(key)
  21         return default
  22     def rec(d, keys):
  23         if len(keys) == 0:
  24             return d
  25         if isinstance(d, dict):
  26             if keys[0] not in d:
  27                 return err(keys[0])
  28             return rec(d[keys[0]], keys[1:])
  29         elif isinstance(d, list):
  30             if not 0 <= keys[0] < len(d):
  31                 return err(keys[0])
  32             return rec(d[keys[0]], keys[1:])
  33         else:
  34             return err(keys[0])
  35     return rec(d, keys)
  36
  37 def linkurl(ln):
  38     if ln[0] != '/':
  39         raise fmterror("unexpected link url: " + ln)
  40     return parse.urljoin(apibase, ln[1:])
  41
  42 def getdsid():
  43     with request.urlopen(loginurl) as resp:
  44         if resp.code != 200:
  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")
  50     return dsel["value"]
  51
  52 def base64(data):
  53     return binascii.b2a_base64(data).decode("ascii").strip().rstrip("=")
  54
  55 class transaction(object):
  56     def __init__(self, account, data):
  57         self.account = account
  58         self._data = data
  59
  60     _datefmt = "%Y-%m-%d"
  61
  62     @property
  63     def value(self): return currency.currency.get(resolve(self._data, ("currency",))).parse(resolve(self._data, ("amount",)))
  64     @property
  65     def message(self): return resolve(self._data, ("description",))
  66     @property
  67     def date(self):
  68         p = time.strptime(resolve(self._data, ("accountingDate",)), self._datefmt)
  69         return datetime.date(p.tm_year, p.tm_mon, p.tm_mday)
  70
  71     @property
  72     def hash(self):
  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()
  79
  80     def __repr__(self):
  81         return "#<fsb.transaction %s: %r>" % (self.value, self.message)
  82
  83 class txnaccount(object):
  84     def __init__(self, sess, id, idata):
  85         self.sess = sess
  86         self.id = id
  87         self._data = None
  88         self._idata = idata
  89
  90     @property
  91     def data(self):
  92         if self._data is None:
  93             self._data = self.sess._jreq("v5/engagement/account/" + self.id)
  94         return self._data
  95
  96     @property
  97     def number(self): return resolve(self.data, ("accountNumber",))
  98     @property
  99     def clearing(self): return resolve(self.data, ("clearingNumber",))
 100     @property
 101     def fullnumber(self): return resolve(self.data, ("fullyFormattedNumber",))
 102     @property
 103     def balance(self): return currency.currency.get(resolve(self.data, ("balance", "currencyCode"))).parse(resolve(self.data, ("balance", "amount")))
 104     @property
 105     def name(self): return resolve(self._idata, ("name",))
 106
 107     def transactions(self):
 108         pagesz = 50
 109         page = 1
 110         while True:
 111             data = self.sess._jreq("v5/engagement/transactions/" + self.id, transactionsPerPage=pagesz, page=page)
 112             txlist = resolve(data, ("transactions",))
 113             if len(txlist) < 1:
 114                 break
 115             for tx in txlist:
 116                 yield transaction(self, tx)
 117             page += 1
 118
 119     def __repr__(self):
 120         return "#<fsb.txnaccount %s: %r>" % (self.fullnumber, self.name)
 121
 122 class cardtransaction(object):
 123     def __init__(self, account, data):
 124         self.account = account
 125         self._data = data
 126
 127     _datefmt = "%Y-%m-%d"
 128
 129     @property
 130     def value(self):
 131         am = resolve(self._data, ("localAmount",))
 132         return currency.currency.get(resolve(am, ("currencyCode",))).parse(resolve(am, ("amount",)))
 133     @property
 134     def message(self): return resolve(self._data, ("description",))
 135     @property
 136     def date(self):
 137         p = time.strptime(resolve(self._data, ("date",)), self._datefmt)
 138         return datetime.date(p.tm_year, p.tm_mon, p.tm_mday)
 139
 140     @property
 141     def hash(self):
 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()
 148
 149     def __repr__(self):
 150         return "#<fsb.cardtransaction %s: %r>" % (self.value, self.message)
 151
 152 class cardaccount(object):
 153     def __init__(self, sess, id, idata):
 154         self.sess = sess
 155         self.id = id
 156         self._data = None
 157         self._idata = idata
 158
 159     @property
 160     def data(self):
 161         if self._data is None:
 162             self._data = self.sess._jreq("v5/engagement/cardaccount/" + self.id)
 163         return self._data
 164
 165     @property
 166     def number(self): return resolve(self.data, ("cardAccount", "cardNumber"))
 167     @property
 168     def balance(self):
 169         cc = resolve(self.data, ("transactions", 0, "localAmount", "currencyCode"))
 170         return currency.currency.get(cc).parse(resolve(self.data, ("cardAccount", "currentBalance")))
 171     @property
 172     def name(self): return resolve(self._idata, ("name",))
 173
 174     def transactions(self):
 175         pagesz = 50
 176         page = 1
 177         while True:
 178             data = self.sess._jreq("v5/engagement/cardaccount/" + self.id, transactionsPerPage=pagesz, page=page)
 179             txlist = resolve(data, ("transactions",))
 180             if len(txlist) < 1:
 181                 break
 182             for tx in txlist:
 183                 yield cardtransaction(self, tx)
 184             page += 1
 185
 186     def __repr__(self):
 187         return "#<fsb.cardaccount %s: %r>" % (self.fullnumber, self.name)
 188
 189 class session(object):
 190     def __init__(self, dsid):
 191         self.dsid = 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))
 200         self.userid = None
 201         self._accounts = None
 202
 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)
 222             return resp.read()
 223
 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"))
 229
 230     def _postlogin(self):
 231         auth = self._jreq("v5/user/authenticationinfo")
 232         uid = auth.get("identifiedUser", "")
 233         if uid == "":
 234             raise fmterror("no identified user even after successful authentication")
 235         self.userid = uid
 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")
 241
 242     def auth_bankid(self, user, conv=None):
 243         if conv is None:
 244             conv = auth.default()
 245         data = self._jreq("v5/identification/bankid/mobile", data = {
 246             "userId": user,
 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")))
 252         fst = None
 253         while True:
 254             time.sleep(3)
 255             vdat = self._jreq(vfy)
 256             st = vdat.get("status")
 257             if st in {"USER_SIGN", "CLIENT_NOT_STARTED"}:
 258                 if st != fst:
 259                     conv.message("Status: %s" % (st,), auth.conv.msg_info)
 260                     fst = st
 261                 continue
 262             elif st == "COMPLETE":
 263                 self._postlogin()
 264                 return
 265             elif st == "CANCELLED":
 266                 raise autherror("authentication cancelled")
 267             else:
 268                 raise fmterror("unexpected bankid status: " + str(st))
 269
 270     def keepalive(self):
 271         data = self._jreq("v5/framework/clientsession")
 272         return data["timeoutInMillis"] / 1000
 273
 274     @property
 275     def accounts(self):
 276         if self._accounts is None:
 277             data = self._jreq("v5/engagement/overview")
 278             accounts = []
 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
 285
 286     def logout(self):
 287         if self.userid is not None:
 288             self._jreq("v5/identification/logout", method="PUT")
 289             self.userid = None
 290
 291     def close(self):
 292         self.logout()
 293         self._req("v5/framework/clientsession", method="DELETE")
 294
 295     def __enter__(self):
 296         return self
 297
 298     def __exit__(self, *excinfo):
 299         self.close()
 300         return False
 301
 302     def __repr__(self):
 303         if self.userid is not None:
 304             return "#<fsb.session %s>" % self.userid
 305         return "#<fsb.session>"
 306
 307     @classmethod
 308     def create(cls):
 309         return cls(getdsid())
 310
 311     def save(self, filename):
 312         with open(filename, "wb") as fp:
 313             pickle.dump(self, fp)
 314
 315     @classmethod
 316     def load(cls, filename):
 317         with open(filename, "rb") as fp:
 318             return pickle.load(fp)