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
   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():
  19         if default is fmterror:
  20             raise fmterror()
  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()
  28             return rec(d[keys[0]], keys[1:])
  29         else:
  30             return err()
  31     return rec(d, keys)
  32
  33 def linkurl(ln):
  34     if ln[0] != '/':
  35         raise fmterror("unexpected link url: " + ln)
  36     return parse.urljoin(apibase, ln[1:])
  37
  38 def getdsid():
  39     with request.urlopen(loginurl) as resp:
  40         if resp.code != 200:
  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")
  46     return dsel["value"]
  47
  48 def base64(data):
  49     return binascii.b2a_base64(data).decode("ascii").strip().rstrip("=")
  50
  51 class transaction(object):
  52     def __init__(self, account, data):
  53         self.account = account
  54         self._data = data
  55
  56     _datefmt = "%Y-%m-%d"
  57
  58     @property
  59     def value(self): return currency.currency.get(resolve(self._data, ("currency",))).parse(resolve(self._data, ("amount",)))
  60     @property
  61     def message(self): return resolve(self._data, ("description",))
  62     @property
  63     def date(self):
  64         p = time.strptime(resolve(self._data, ("accountingDate",)), self._datefmt)
  65         return datetime.date(p.tm_year, p.tm_mon, p.tm_mday)
  66
  67     @property
  68     def hash(self):
  69         dig = hashlib.sha256()
  70         dig.update(str(self.date.toordinal()).encode("ascii") + b"\0")
  71         dig.update(self.message.encode("utf-8") + b"\0")
  72         dig.update(str(self.value.amount).encode("ascii") + b"\0")
  73         dig.update(self.value.currency.symbol.encode("ascii") + b"\0")
  74         return dig.hexdigest()
  75
  76     def __repr__(self):
  77         return "#<fsb.transaction %s: %r>" % (self.value, self.message)
  78
  79 class account(object):
  80     def __init__(self, sess, id, idata):
  81         self.sess = sess
  82         self.id = id
  83         self._data = None
  84         self._idata = idata
  85
  86     @property
  87     def data(self):
  88         if self._data is None:
  89             self._data = self.sess._jreq("v5/engagement/account/" + self.id)
  90         return self._data
  91
  92     @property
  93     def number(self): return resolve(self.data, ("accountNumber",))
  94     @property
  95     def clearing(self): return resolve(self.data, ("clearingNumber",))
  96     @property
  97     def fullnumber(self): return resolve(self.data, ("fullyFormattedNumber",))
  98     @property
  99     def balance(self): return currency.currency.get(resolve(self.data, ("balance", "currencyCode"))).parse(resolve(self.data, ("balance", "amount")))
 100     @property
 101     def name(self): return resolve(self._idata, ("name",))
 102
 103     def transactions(self):
 104         pagesz = 50
 105         page = 1
 106         while True:
 107             data = self.sess._jreq("v5/engagement/transactions/" + self.id, transactionsPerPage=pagesz, page=page)
 108             txlist = resolve(data, ("transactions",))
 109             if len(txlist) < 1:
 110                 break
 111             for tx in txlist:
 112                 yield transaction(self, tx)
 113             page += 1
 114
 115     def __repr__(self):
 116         return "#<fsb.account %s: %r>" % (self.fullnumber, self.name)
 117
 118 class session(object):
 119     def __init__(self, dsid):
 120         self.dsid = dsid
 121         self.auth = base64((serviceid + ":" + str(int(time.time() * 1000))).encode("ascii"))
 122         self.jar = request.HTTPCookieProcessor()
 123         self.jar.cookiejar.set_cookie(http.cookiejar.Cookie(
 124             version=0, name="dsid", value=dsid, path="/", path_specified=True,
 125             domain=".online.swedbank.se", domain_specified=True, domain_initial_dot=True,
 126             port=None, port_specified=False, secure=False, expires=None,
 127             discard=True, comment=None, comment_url=None,
 128             rest={}, rfc2109=False))
 129         self.userid = None
 130         self._accounts = None
 131
 132     def _req(self, url, data=None, ctype=None, headers={}, method=None, **kws):
 133         if "dsid" not in kws:
 134             kws["dsid"] = self.dsid
 135         kws = {k: v for (k, v) in kws.items() if v is not None}
 136         url = parse.urljoin(apibase, url + "?" + parse.urlencode(kws))
 137         if isinstance(data, dict):
 138             data = json.dumps(data).encode("utf-8")
 139             ctype = "application/json;charset=UTF-8"
 140         req = request.Request(url, data=data, method=method)
 141         for hnam, hval in headers.items():
 142             req.add_header(hnam, hval)
 143         if ctype is not None:
 144             req.add_header("Content-Type", ctype)
 145         req.add_header("Authorization", self.auth)
 146         self.jar.https_request(req)
 147         with request.urlopen(req) as resp:
 148             if resp.code != 200 and resp.code != 201:
 149                 raise fmterror("Unexpected HTTP status code: " + str(resp.code))
 150             self.jar.https_response(req, resp)
 151             return resp.read()
 152
 153     def _jreq(self, *args, **kwargs):
 154         headers = kwargs.pop("headers", {})
 155         headers["Accept"] = "application/json"
 156         ret = self._req(*args, headers=headers, **kwargs)
 157         return json.loads(ret.decode("utf-8"))
 158
 159     def _postlogin(self):
 160         auth = self._jreq("v5/user/authenticationinfo")
 161         uid = auth.get("identifiedUser", "")
 162         if uid == "":
 163             raise fmterror("no identified user even after successful authentication")
 164         self.userid = uid
 165         prof = self._jreq("v5/profile/")
 166         if len(prof["banks"]) != 1:
 167             raise fmterror("do not know the meaning of multiple banks")
 168         rolesw = linkurl(resolve(prof["banks"][0], ("privateProfile", "links", "next", "uri")))
 169         self._jreq(rolesw, method="POST")
 170
 171     def auth_bankid(self, user):
 172         data = self._jreq("v5/identification/bankid/mobile", data = {
 173             "userId": user,
 174             "useEasyLogin": False,
 175             "generateEasyLoginId": False})
 176         if data.get("status") != "USER_SIGN":
 177             raise fmterror("unexpected bankid status: " + str(data.get("status")))
 178         vfy = linkurl(resolve(data, ("links", "next", "uri")))
 179         while True:
 180             time.sleep(3)
 181             vdat = self._jreq(vfy)
 182             st = vdat.get("status")
 183             if st == "USER_SIGN":
 184                 continue
 185             elif st == "CLIENT_NOT_STARTED":
 186                 continue
 187             elif st == "COMPLETE":
 188                 self._postlogin()
 189                 return
 190             elif st == "CANCELLED":
 191                 raise autherror("authentication cancelled")
 192             else:
 193                 raise fmterror("unexpected bankid status: " + str(st))
 194
 195     def keepalive(self):
 196         data = self._jreq("v5/framework/clientsession")
 197         return data["timeoutInMillis"] / 1000
 198
 199     @property
 200     def accounts(self):
 201         if self._accounts is None:
 202             data = self._jreq("v5/engagement/overview")
 203             accounts = []
 204             for acct in resolve(data, ("transactionAccounts",)):
 205                 accounts.append(account(self, resolve(acct, ("id",)), acct))
 206             self._accounts = accounts
 207         return self._accounts
 208
 209     def logout(self):
 210         if self.userid is not None:
 211             self._jreq("v5/identification/logout", method="PUT")
 212             self.userid = None
 213
 214     def close(self):
 215         self.logout()
 216         self._req("v5/framework/clientsession", method="DELETE")
 217
 218     def __enter__(self):
 219         return self
 220
 221     def __exit__(self, *excinfo):
 222         self.close()
 223         return False
 224
 225     def __repr__(self):
 226         if self.userid is not None:
 227             return "#<fsb.session %s>" % self.userid
 228         return "#<fsb.session>"
 229
 230     @classmethod
 231     def create(cls):
 232         return cls(getdsid())
 233
 234     def save(self, filename):
 235         with open(filename, "wb") as fp:
 236             pickle.dump(self, fp)
 237
 238     @classmethod
 239     def load(cls, filename):
 240         with open(filename, "rb") as fp:
 241             return picke.load(fp)