-import json, http.cookiejar, binascii, time, datetime, pickle, hashlib
+import json, http.cookiejar, binascii, time, datetime, pickle, urllib.error, io
+from PIL import Image
from urllib import request, parse
from bs4 import BeautifulSoup as soup
-from . import currency, auth
+from . import currency, auth, data
soupify = lambda cont: soup(cont, "html.parser")
apibase = "https://online.swedbank.se/TDE_DAP_Portal_REST_WEB/api/"
class fmterror(Exception):
pass
-class autherror(Exception):
+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:
def base64(data):
return binascii.b2a_base64(data).decode("ascii").strip().rstrip("=")
-class transaction(object):
+class transaction(data.transaction):
def __init__(self, account, data):
self.account = account
self._data = data
p = time.strptime(resolve(self._data, ("accountingDate",)), self._datefmt)
return datetime.date(p.tm_year, p.tm_mon, p.tm_mday)
- @property
- def hash(self):
- dig = hashlib.sha256()
- dig.update(str(self.date.toordinal()).encode("ascii") + b"\0")
- dig.update(self.message.encode("utf-8") + b"\0")
- dig.update(str(self.value.amount).encode("ascii") + b"\0")
- dig.update(self.value.currency.symbol.encode("ascii") + b"\0")
- return dig.hexdigest()
-
- def __repr__(self):
- return "#<fsb.transaction %s: %r>" % (self.value, self.message)
-
-class txnaccount(object):
+class txnaccount(data.txnaccount):
def __init__(self, sess, id, idata):
self.sess = sess
self.id = id
yield transaction(self, tx)
page += 1
- def __repr__(self):
- return "#<fsb.txnaccount %s: %r>" % (self.fullnumber, self.name)
-
-class cardtransaction(object):
+class cardtransaction(data.transaction):
def __init__(self, account, data):
self.account = account
self._data = data
p = time.strptime(resolve(self._data, ("date",)), self._datefmt)
return datetime.date(p.tm_year, p.tm_mon, p.tm_mday)
- @property
- def hash(self):
- dig = hashlib.sha256()
- dig.update(str(self.date.toordinal()).encode("ascii") + b"\0")
- dig.update(self.message.encode("utf-8") + b"\0")
- dig.update(str(self.value.amount).encode("ascii") + b"\0")
- dig.update(self.value.currency.symbol.encode("ascii") + b"\0")
- return dig.hexdigest()
-
- def __repr__(self):
- return "#<fsb.cardtransaction %s: %r>" % (self.value, self.message)
-
-class cardaccount(object):
+class cardaccount(data.cardaccount):
def __init__(self, sess, id, idata):
self.sess = sess
self.id = id
yield cardtransaction(self, tx)
page += 1
- def __repr__(self):
- return "#<fsb.cardaccount %s: %r>" % (self.fullnumber, self.name)
-
-class session(object):
+class session(data.session):
def __init__(self, dsid):
self.dsid = dsid
self.auth = base64((serviceid + ":" + str(int(time.time() * 1000))).encode("ascii"))
def _jreq(self, *args, **kwargs):
headers = kwargs.pop("headers", {})
headers["Accept"] = "application/json"
- ret = self._req(*args, headers=headers, **kwargs)
+ 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):
rolesw = linkurl(resolve(prof["banks"][0], ("privateProfile", "links", "next", "uri")))
self._jreq(rolesw, method="POST")
+ def auth_token(self, user, conv=None):
+ if conv is None:
+ conv = auth.default()
+ try:
+ data = self._jreq("v5/identification/securitytoken/challenge", 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
+ if data.get("useOneTimePassword"):
+ raise fmterror("unexpectedly found useOneTimePassword")
+ if data.get("challenge") != "":
+ raise fmterror("unexpected challenge: " + str(data.get("challenge")))
+ if not isinstance(data.get("imageChallenge"), dict) or resolve(data, ("imageChallenge", "method")) != "GET":
+ raise fmterror("invalid image challenge: " + str(data.get("imageChallenge")))
+ iurl = linkurl(resolve(data, ("imageChallenge", "uri")))
+ vfy = linkurl(resolve(data, ("links", "next", "uri")))
+ img = Image.open(io.BytesIO(self._req(iurl)))
+ conv.image(img)
+ response = conv.prompt("Token response: ", True)
+ try:
+ data = self._jreq(vfy, data={"response": response})
+ except jsonerror as e:
+ msgs = resolve(e.data, ("errorMessages", "general"), False)
+ if isinstance(msgs, list):
+ for msg in msgs:
+ if msg.get("message"):
+ raise autherror(msg.get("message"))
+ raise
+ if not data.get("authenticationRole", ""):
+ raise fmterror("authentication appears to have succeded, but there is no authenticationRole: " + str(data))
+ self._postlogin()
+
def auth_bankid(self, user, conv=None):
if conv is None:
conv = auth.default()
- data = self._jreq("v5/identification/bankid/mobile", data = {
- "userId": user,
- "useEasyLogin": False,
- "generateEasyLoginId": False})
- if data.get("status") != "USER_SIGN":
- raise fmterror("unexpected bankid status: " + str(data.get("status")))
+ 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:
- time.sleep(3)
- vdat = self._jreq(vfy)
- st = vdat.get("status")
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")
@property
def accounts(self):
if self._accounts is None:
- data = self._jreq("v5/engagement/overview")
+ txndata = self._jreq("v5/engagement/overview")
+ crddata = self._jreq("v5/card/creditcard")
accounts = []
- for acct in resolve(data, ("transactionAccounts",)):
+ for acct in resolve(txndata, ("transactionAccounts",)):
accounts.append(txnaccount(self, resolve(acct, ("id",)), acct))
- for acct in resolve(data, ("cardAccounts",)):
+ for acct in resolve(crddata, ("cardAccounts",)):
accounts.append(cardaccount(self, resolve(acct, ("id",)), acct))
self._accounts = accounts
return self._accounts
self.logout()
self._req("v5/framework/clientsession", method="DELETE")
+ def __getstate__(self):
+ state = dict(self.__dict__)
+ state["jar"] = list(state["jar"].cookiejar)
+ return state
+
+ def __setstate__(self, state):
+ jar = request.HTTPCookieProcessor()
+ for cookie in state["jar"]:
+ jar.cookiejar.set_cookie(cookie)
+ state["jar"] = jar
+ self.__dict__.update(state)
+
def __enter__(self):
return self
@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)
Dolda2000 GitWeb / fulbank.git / blobdiff