Compare commits

..

No commits in common. "master" and "lash/settable-meta-origin" have entirely different histories.

16 changed files with 177 additions and 498 deletions

4
.gitignore vendored
View File

@ -2,7 +2,3 @@ __pycache__
*.egg-info
build/
*.pyc
.venv
.clicada
dist/
.vscode/

View File

@ -1,14 +1,3 @@
- 0.0.7
* fix: make store_path relative to the users home
- 0.0.6
* Add cache encryption, with AES-CTR-128
- 0.0.5
* Replace logs with colorized progress output on default loglevel
* Do not repeat already failed metadata lookups
- 0.0.4
* Resolve metadata to labels when loading from cache
- 0.0.3
* Upgrade usumbufu to prevent missing bearer auth on https
- 0.0.2
* Use ~/.config for default config override
- 0.0.1-unreleased

View File

@ -1,87 +0,0 @@
## Clicada
> Admin Command Line Interface to interact with cic-meta and cic-cache
### Pre-requisites
- Public key uploaded to `cic-auth-helper`
- PGP Keyring for your key
### Installation
Use either of the following installation methods:
1. Install from git release (recommended)
```bash
wget https://git.grassecon.net/grassrootseconomics/clicada/archive/v0.0.6.zip
unzip clicada-v0.0.6.zip
cd clicada
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt --extra-index-url=https://pip.grassrootseconomics.net
```
2. Install from pip to path (non sudo)
```bash
pip3 install -UI --extra-index-url=https://pip.grassrootseconomics.net clicada
```
### GPG Keyring setup
PGP uses the default keyring, you can however pass in a custom keyring path.
To create a keyring from a specific key and get its path for `AUTH_KEYRING_PATH`:
```bash
# In some dir
gpg --homedir .gnupg --import private.pgp
pwd
```
### Usage
```bash
usage: clicada [...optional arguments] [...positional arguments]
positional arguments:
{user,u,tag,t}
user (u) retrieve transactions for a user
tag (t) locally assign a display value to an identifier
optional arguments:
-h, --help show this help message and exit
--no-logs Turn off all logging
-v Be verbose
-vv Be very verbose
-c CONFIG, --config CONFIG
Configuration directory
-n NAMESPACE, --namespace NAMESPACE
Configuration namespace
--dumpconfig {env,ini}
Output configuration and quit. Use with --raw to omit values and output schema only.
--env-prefix ENV_PREFIX
environment prefix for variables to overwrite configuration
-p P, --rpc-provider P
RPC HTTP(S) provider url
--rpc-dialect RPC_DIALECT
RPC HTTP(S) backend dialect
--height HEIGHT Block height to execute against
-i I, --chain-spec I Chain specification string
-u, --unsafe Do not verify address checksums
--seq Use sequential rpc ids
-y Y, --key-file Y Keystore file to use for signing or address
--raw Do not decode output
--fee-price FEE_PRICE
override fee price
--fee-limit FEE_LIMIT
override fee limit
```
### Example
```bash
AUTH_PASSPHRASE=queenmarlena AUTH_KEYRING_PATH=/home/kamikaze/grassroots/usumbufu/tests/testdata/pgp/.gnupg/ AUTH_KEY=CCE2E1D2D0E36ADE0405E2D0995BB21816313BD5 CHAIN_SPEC=evm:byzantium:8996:bloxberg CIC_REGISTRY_ADDRESS=0xcf60ebc445b636a5ab787f9e8bc465a2a3ef8299 RPC_PROVIDER=https://rpc.grassecon.net TX_CACHE_URL=https://cache.grassecon.net HTTP_CORS_ORIGIN=https://auth.grassecon.net META_HTTP_ORIGIN=https://auth.grassecon.net:443 PYTHONPATH=. python clicada/runnable/view.py u --meta-url https://auth.grassecon.net +254711000000
```

View File

@ -1,57 +1,40 @@
# import notifier
from clicada.cli.notify import NotifyWriter
notifier = NotifyWriter()
# notifier.notify('loading script')
import importlib
import logging
# standard imports
import os
import sys
import chainlib.eth.cli
import clicada.cli.tag as cmd_tag
# local imports
import clicada.cli.user as cmd_user
#import argparse
import logging
import importlib
# external imports
import confini
import chainlib.eth.cli
from chainlib.chain import ChainSpec
# local imports
import clicada.cli.user as cmd_user
import clicada.cli.tag as cmd_tag
from clicada.cli.auth import PGPAuthCrypt
from clicada.cli.http import HTTPSession, PGPClientSession
from clicada.crypt.aes import AESCTREncrypt
from clicada.cli.http import (
HTTPSession,
PGPClientSession,
)
logg = logging.getLogger()
script_dir = os.path.dirname(os.path.realpath(__file__))
data_dir = os.path.join(script_dir, "..", "data")
base_config_dir = os.path.join(data_dir, "config")
class NullWriter:
def notify(self, v):
pass
def ouch(self, v):
pass
def write(self, v):
sys.stdout.write(str(v))
data_dir = os.path.join(script_dir, '..', 'data')
base_config_dir = os.path.join(data_dir, 'config')
class CmdCtrl:
__cmd_alias = {
"u": "user",
"t": "tag",
}
'u': 'user',
't': 'tag',
}
__auth_for = [
"user",
]
'user',
]
def __init__(self, argv=None, description=None, logger=None, *args, **kwargs):
self.args(argv)
@ -59,46 +42,37 @@ class CmdCtrl:
self.logging(logger)
self.module()
self.config()
self.notifier()
self.auth()
self.blockchain()
self.remote_openers = {}
if self.get("META_URL") != None:
if self.get('META_URL') != None:
auth_client_session = PGPClientSession(self.__auth)
self.remote_openers["meta"] = HTTPSession(
self.get("META_URL"),
auth=auth_client_session,
origin=self.config.get("META_HTTP_ORIGIN"),
)
self.remote_openers['meta'] = HTTPSession(self.get('META_URL'), auth=auth_client_session, origin=self.config.get('META_HTTP_ORIGIN'))
def blockchain(self):
self.chain_spec = ChainSpec.from_chain_str(self.config.get("CHAIN_SPEC"))
self.chain_spec = ChainSpec.from_chain_str(self.config.get('CHAIN_SPEC'))
self.rpc = chainlib.eth.cli.Rpc()
self.__conn = self.rpc.connect_by_config(self.config)
def args(self, argv):
self.argparser = chainlib.eth.cli.ArgumentParser(
chainlib.eth.cli.argflag_std_read
)
self.argparser = chainlib.eth.cli.ArgumentParser(chainlib.eth.cli.argflag_std_read)
sub = self.argparser.add_subparsers()
sub.dest = "command"
sub_user = sub.add_parser(
"user", aliases=["u"], help="retrieve transactions for a user"
)
sub.dest = 'command'
sub_user = sub.add_parser('user', aliases=['u'], help='retrieve transactions for a user')
cmd_user.process_args(sub_user)
sub_tag = sub.add_parser(
"tag", aliases=["t"], help="locally assign a display value to an identifier"
)
sub_tag = sub.add_parser('tag', aliases=['t'], help='locally assign a display value to an identifier')
cmd_tag.process_args(sub_tag)
self.cmd_args = self.argparser.parse_args(argv)
def module(self):
self.cmd_string = self.cmd_args.command
cmd_string_translate = self.__cmd_alias.get(self.cmd_string)
@ -106,12 +80,13 @@ class CmdCtrl:
self.cmd_string = cmd_string_translate
if self.cmd_string == None:
self.cmd_string = "none"
modname = "clicada.cli.{}".format(self.cmd_string)
self.logger.debug("using module {}".format(modname))
self.cmd_string = 'none'
modname = 'clicada.cli.{}'.format(self.cmd_string)
self.logger.debug('using module {}'.format(modname))
self.cmd_mod = importlib.import_module(modname)
def logging(self, logger):
self.logger = logger
if self.logger == None:
@ -120,88 +95,64 @@ class CmdCtrl:
self.logger.setLevel(logging.DEBUG)
elif self.cmd_args.v:
self.logger.setLevel(logging.INFO)
def config(self):
override_dir = self.cmd_args.config
if override_dir == None:
p = os.environ.get("HOME")
p = os.environ.get('HOME')
if p != None:
p = os.path.join(p, ".config", "cic", "clicada")
p = os.path.join(p, '.config', 'cic', 'clicada')
try:
os.stat(p)
override_dir = p
logg.info(
"applying user config override from standard location: {}".format(
p
)
)
logg.info('applying user config override from standard location: {}'.format(p))
except FileNotFoundError:
pass
extra_args = self.cmd_mod.extra_args()
self.config = chainlib.eth.cli.Config.from_args(
self.cmd_args,
base_config_dir=base_config_dir,
extra_args=extra_args,
default_config_dir=override_dir,
)
self.config = chainlib.eth.cli.Config.from_args(self.cmd_args, base_config_dir=base_config_dir, extra_args=extra_args, default_config_dir=override_dir)
self.config.add(False, "_SEQ")
self.config.add(False, '_SEQ')
self.config.censor("AUTH_PASSPHRASE")
self.config.censor('AUTH_PASSPHRASE')
self.logger.debug('loaded config:\n{}'.format(self.config))
self.logger.debug("loaded config:\n{}".format(self.config))
def auth(self):
typ = self.get("AUTH_TYPE")
if typ != "gnupg":
raise NotImplementedError("Valid aut implementations are: gnupg")
typ = self.get('AUTH_TYPE')
if typ != 'gnupg':
raise NotImplementedError('Valid aut implementations are: gnupg')
default_auth_db_path = None
if os.environ.get("HOME") != None:
default_auth_db_path = os.path.join(
os.environ["HOME"], ".local/share/cic/clicada"
)
auth_db_path = self.get("AUTH_DB_PATH", default_auth_db_path)
self.__auth = PGPAuthCrypt(
auth_db_path, self.get("AUTH_KEY"), self.get("AUTH_KEYRING_PATH")
)
self.__auth.get_secret(self.get("AUTH_PASSPHRASE"))
self.encrypter = AESCTREncrypt(auth_db_path, self.__auth.secret)
if os.environ.get('HOME') != None:
default_auth_db_path = os.path.join(os.environ['HOME'], '.local/share/cic/clicada')
auth_db_path = self.get('AUTH_DB_PATH', default_auth_db_path)
self.__auth = PGPAuthCrypt(auth_db_path, self.get('AUTH_KEY'), self.get('AUTH_KEYRING_PATH'))
self.__auth.get_secret(self.get('AUTH_PASSPHRASE'))
def get(self, k, default=None):
r = self.config.get(k, default)
if k in [
"_FORCE",
]:
'_FORCE',
]:
if r == None:
return False
return self.config.true(k)
return r
def chain(self):
return self.chain_spec
def conn(self):
return self.__conn
def execute(self):
self.cmd_mod.execute(self)
def opener(self, k):
return self.remote_openers[k]
def notifier(self):
if logg.root.level >= logging.WARNING:
logging.disable()
self.writer = notifier
else:
self.writer = NullWriter()
def notify(self, v):
if logg.root.level <= logging.INFO:
print("\033[96m" + v + "\033[0m")
def ouch(self, v):
print("\033[91m" + v + "\033[0m")
def write(self, v):
print(v)

View File

@ -26,7 +26,6 @@ class PGPAuthCrypt:
raise AuthError('invalid key {}'.format(auth_key))
self.auth_key = auth_key
self.gpg = gnupg.GPG(gnupghome=pgp_dir)
self.secret = None
def get_secret(self, passphrase=''):
@ -50,11 +49,10 @@ class PGPAuthCrypt:
f.write(secret.data)
f.close()
f = open(p, 'rb')
secret = self.gpg.decrypt_file(f, passphrase=passphrase)
if not secret.ok:
self.secret = self.gpg.decrypt_file(f, passphrase=passphrase)
if not self.secret.ok:
raise AuthError('could not decrypt encryption secret. wrong password?')
f.close()
self.secret = secret.data
self.__passphrase = passphrase

View File

@ -3,7 +3,6 @@ import hashlib
import urllib.parse
import os
import logging
from socket import getservbyname
# external imports
from usumbufu.client.base import (
@ -49,15 +48,7 @@ class HTTPSession:
def __init__(self, url, auth=None, origin=None):
self.base_url = url
url_parts = urllib.parse.urlsplit(self.base_url)
url_parts_origin_host = url_parts[1].split(":")
host = url_parts_origin_host[0]
try:
host = host + ':' + url_parts_origin_host[1]
except IndexError:
host = host + ':' + str(getservbyname(url_parts[0]))
logg.info('changed origin with missing port number from {} to {}'.format(url_parts[1], host))
url_parts_origin = (url_parts[0], host, '', '', '',)
url_parts_origin = (url_parts[0], url_parts[1], '', '', '',)
self.origin = origin
if self.origin == None:
self.origin = urllib.parse.urlunsplit(url_parts_origin)

View File

@ -1,33 +0,0 @@
# standard imports
import os
import sys
import shutil
class NotifyWriter:
def __init__(self, writer=sys.stdout):
(c, r) = shutil.get_terminal_size()
self.cols = c
self.fmt = "\r{:" + "<{}".format(c) + "}"
self.w = writer
self.notify_max = self.cols - 4
def notify(self, v):
if len(v) > self.notify_max:
v = v[:self.notify_max]
self.write('\x1b[0;36m... ' + v + '\x1b[0;39m')
def ouch(self, v):
if len(v) > self.notify_max:
v = v[:self.notify_max]
self.write('\x1b[0;91m!!! ' + v + '\x1b[0;39m')
def write(self, v):
s = str(v)
if len(s) > self.cols:
s = s[:self.cols]
self.w.write(self.fmt.format(s))

View File

@ -3,8 +3,7 @@ import json
# external imports
from clicada.user import FileUserStore
from pathlib import Path
import os
categories = [
'phone',
@ -36,7 +35,7 @@ def validate(config, args):
def execute(ctrl):
store_path = os.path.join(str(Path.home()), '.clicada')
store_path = '.clicada'
user_store = FileUserStore(None, ctrl.chain(), ctrl.get('_CATEGORY'), store_path, int(ctrl.get('FILESTORE_TTL')))
user_store.put(ctrl.get('_IDENTIFIER'), json.dumps(ctrl.get('_TAG')), force=True)
user_store.stick(ctrl.get('_IDENTIFIER'))

View File

@ -1,24 +1,24 @@
# standard imports
import datetime
import logging
import os
import sys
from pathlib import Path
from chainlib.encode import TxHexNormalizer
from chainlib.eth.address import is_address, to_checksum_address
import logging
import datetime
# external imports
from cic_eth_registry import CICRegistry
from cic_eth_registry.lookup.tokenindex import TokenIndexLookup
from cic_types.models.person import Person
from clicada.error import MetadataNotFoundError
from clicada.token import FileTokenStore, token_balance
from chainlib.eth.address import to_checksum_address
from chainlib.encode import TxHexNormalizer
from hexathon import add_0x
# local imports
from clicada.tx import ResolvedTokenTx, TxGetter
from clicada.tx import TxGetter
from clicada.user import FileUserStore
from hexathon import add_0x
from clicada.token import (
FileTokenStore,
token_balance,
)
from clicada.tx import ResolvedTokenTx
logg = logging.getLogger(__name__)
@ -26,36 +26,24 @@ tx_normalizer = TxHexNormalizer()
def process_args(argparser):
argparser.add_argument("-m", "--method", type=str, help="lookup method")
argparser.add_argument(
"--meta-url", dest="meta_url", type=str, help="Url to retrieve metadata from"
)
argparser.add_argument(
"-f",
"--force-update",
dest="force_update",
action="store_true",
help="Update records of mutable entries",
)
argparser.add_argument(
"identifier", type=str, help="user identifier (phone_number or address)"
)
argparser.add_argument('-m', '--method', type=str, help='lookup method')
argparser.add_argument('--meta-url', dest='meta_url', type=str, help='Url to retrieve metadata from')
argparser.add_argument('-f', '--force-update', dest='force_update', action='store_true', help='Update records of mutable entries')
argparser.add_argument('identifier', type=str, help='user identifier')
def extra_args():
return {
"force_update": "_FORCE",
"method": "META_LOOKUP_METHOD",
"meta_url": "META_URL",
"identifier": "_IDENTIFIER",
}
'force_update': '_FORCE',
'method': 'META_LOOKUP_METHOD',
'meta_url': 'META_URL',
'identifier': '_IDENTIFIER',
}
def apply_args(config, args):
if config.get("META_LOOKUP_METHOD"):
raise NotImplementedError(
'Sorry, currently only "phone" lookup method is implemented'
)
if config.get('META_LOOKUP_METHOD'):
raise NotImplementedError('Sorry, currently only "phone" lookup method is implemented')
def validate(config, args):
@ -63,103 +51,74 @@ def validate(config, args):
def execute(ctrl):
tx_getter = TxGetter(ctrl.get("TX_CACHE_URL"), 10)
tx_getter = TxGetter(ctrl.get('TX_CACHE_URL'))
store_path = os.path.join(str(Path.home()), ".clicada")
user_phone_file_label = "phone"
user_phone_store = FileUserStore(
ctrl.opener("meta"),
ctrl.chain(),
user_phone_file_label,
store_path,
int(ctrl.get("FILESTORE_TTL")),
encrypter=ctrl.encrypter,
)
store_path = '.clicada'
user_phone_file_label = 'phone'
user_phone_store = FileUserStore(ctrl.opener('meta'), ctrl.chain(), user_phone_file_label, store_path, int(ctrl.get('FILESTORE_TTL')))
identifier = ctrl.get("_IDENTIFIER")
ctrl.notify("resolving identifier {} to wallet address".format(identifier))
if is_address(identifier):
user_address = identifier
else:
user_address = user_phone_store.by_phone(identifier, update=ctrl.get("_FORCE"))
user_address = user_phone_store.by_phone(ctrl.get('_IDENTIFIER'), update=ctrl.get('_FORCE'))
if user_address == None:
ctrl.ouch("unknown identifier: {}\n".format(identifier))
sys.stderr.write('unknown identifier: {}\n'.format(ctrl.get('_IDENTIFIER')))
sys.exit(1)
try:
user_address = to_checksum_address(user_address)
except ValueError:
ctrl.ouch('invalid response "{}" for {}\n'.format(user_address, identifier))
sys.stderr.write('invalid response "{}" for {}\n'.format(user_address, ctrl.get('_IDENTIFIER')))
sys.exit(1)
logg.debug("loaded user address {} for {}".format(user_address, identifier))
logg.debug('loaded user address {} for {}'.format(user_address, ctrl.get('_IDENTIFIER')))
user_address_normal = tx_normalizer.wallet_address(user_address)
ctrl.notify("retrieving txs for address {}".format(user_address_normal))
txs = tx_getter.get(user_address)
token_store = FileTokenStore(ctrl.chain(), ctrl.conn(), "token", store_path)
token_store = FileTokenStore(ctrl.chain(), ctrl.conn(), 'token', store_path)
user_address_file_label = "address"
user_address_store = FileUserStore(
ctrl.opener("meta"),
ctrl.chain(),
user_address_file_label,
store_path,
int(ctrl.get("FILESTORE_TTL")),
encrypter=ctrl.encrypter,
)
user_address_file_label = 'address'
user_address_store = FileUserStore(ctrl.opener('meta'), ctrl.chain(), user_address_file_label, store_path, int(ctrl.get('FILESTORE_TTL')))
r = None
ctrl.write(
f"""
Phone: {ctrl.get("_IDENTIFIER")}
Network address: {add_0x(user_address)}
Chain: {ctrl.chain().common_name()}"""
user_address_normal = tx_normalizer.wallet_address(user_address)
r = user_address_store.by_address(user_address_normal, update=ctrl.get('_FORCE'))
print('r {}'.format(r))
print("""Phone: {}
Network address: {}
Chain: {}
Name: {}
Registered: {}
Gender: {}
Location: {}
Products: {}
Tags: {}
Balances:""".format(
ctrl.get('_IDENTIFIER'),
add_0x(user_address),
ctrl.chain().common_name(),
str(r),
datetime.datetime.fromtimestamp(r.date_registered).ctime(),
r.gender,
r.location['area_name'],
','.join(r.products),
','.join(r.tags),
)
ctrl.notify("resolving metadata for address {}".format(user_address_normal))
try:
r = user_address_store.by_address(
user_address_normal, update=ctrl.get("_FORCE")
)
if r:
ctrl.write(
f"""
Name: { str(r)}
Registered: {datetime.datetime.fromtimestamp(r.date_registered).ctime()}
Gender: {r.gender}
Location: {r.location["area_name"]}
Products: {",".join(r.products)}
Tags: {",".join(r.tags)}"""
)
except MetadataNotFoundError as e:
ctrl.ouch(f"MetadataNotFoundError: Could not resolve metadata for user {e}\n")
)
tx_lines = []
seen_tokens = {}
for tx_src in txs["data"]:
ctrl.notify("resolve details for tx {}".format(tx_src["tx_hash"]))
for tx_src in txs['data']:
tx = ResolvedTokenTx.from_dict(tx_src)
tx.resolve(
token_store,
user_address_store,
show_decimals=True,
update=ctrl.get("_FORCE"),
)
tx.resolve(token_store, user_address_store, show_decimals=True, update=ctrl.get('_FORCE'))
tx_lines.append(tx)
seen_tokens[tx.source_token_label] = tx.source_token
seen_tokens[tx.destination_token_label] = tx.destination_token
for k in seen_tokens.keys():
ctrl.notify("resolve token {}".format(seen_tokens[k]))
(token_symbol, token_decimals) = token_store.by_address(seen_tokens[k])
ctrl.notify(
"get token balance for {} => {}".format(token_symbol, seen_tokens[k])
)
balance = token_balance(ctrl.chain(), ctrl.conn(), seen_tokens[k], user_address)
fmt = "{:." + str(token_decimals) + "f}"
decimal_balance = fmt.format(balance / (10**token_decimals))
ctrl.write("Balances:\n {} {}".format(token_symbol, decimal_balance))
fmt = '{:.' + str(token_decimals) + 'f}'
decimal_balance = fmt.format(balance / (10 ** token_decimals))
print("\t{} {}".format(token_symbol, decimal_balance))
print()
for l in tx_lines:
ctrl.write(l)
print(l)

View File

@ -1,42 +0,0 @@
# standard imports
import os
import logging
import hashlib
from Crypto.Cipher import AES
from Crypto.Util import Counter
from .base import Encrypter
logg = logging.getLogger(__name__)
class AESCTREncrypt(Encrypter):
aes_block_size = 1 << 7
counter_bytes = int(128 / 8)
def __init__(self, db_dir, secret):
self.secret = secret
def key_to_iv(self, k):
h = hashlib.sha256()
h.update(k.encode('utf-8'))
h.update(self.secret)
z = h.digest()
return int.from_bytes(z[:self.counter_bytes], 'big')
def encrypt(self, k, v):
iv = self.key_to_iv(k)
ctr = Counter.new(self.aes_block_size, initial_value=iv)
cipher = AES.new(self.secret, AES.MODE_CTR, counter=ctr)
return cipher.encrypt(v)
def decrypt(self, k, v):
iv = self.key_to_iv(k)
ctr = Counter.new(self.aes_block_size, initial_value=iv)
cipher = AES.new(self.secret, AES.MODE_CTR, counter=ctr)
return cipher.decrypt(v)

View File

@ -1,8 +0,0 @@
class Encrypter:
def encrypt(self, v):
raise NotImplementedError()
def decrypt(self, v):
raise NotImplementedError()

View File

@ -4,7 +4,3 @@ class ExpiredRecordError(Exception):
class AuthError(Exception):
pass
class MetadataNotFoundError(Exception):
pass

View File

@ -12,10 +12,7 @@ from cic_types.models.tx import (
# local imports
from clicada.encode import tx_normalize
from clicada.error import (
ExpiredRecordError,
MetadataNotFoundError,
)
from clicada.error import ExpiredRecordError
logg = logging.getLogger(__name__)
@ -59,22 +56,35 @@ class ResolvedTokenTx(TokenTx):
self.to_value_label = fmt.format(token_value)
def resolve_entity(self, user_store, address):
def resolve_stored_entity(self, user_store, address, update=False):
if update:
return None
address = tx_normalize.wallet_address(address)
try:
r = user_store.by_address(address)
except MetadataNotFoundError:
return address
return str(r)
v = user_store.get(address)
return v
except FileNotFoundError:
return None
except ExpiredRecordError:
return None
def resolve_sender_entity(self, user_store, update=False):
v = self.resolve_stored_entity(user_store, self.sender, update=update)
if v != None:
return v
if self.tx_type == TokenTxType.faucet_giveto.value:
return 'FAUCET'
return self.resolve_entity(user_store, self.sender)
r = user_store.by_address(self.sender)
return str(r)
def resolve_recipient_entity(self, user_store, update=False):
return self.resolve_entity(user_store, self.recipient)
v = self.resolve_stored_entity(user_store, self.recipient, update=update)
if v != None:
return v
r = user_store.by_address(self.recipient, update=update)
return str(r)
def resolve_entities(self, user_store, update=False):
@ -89,7 +99,7 @@ class ResolvedTokenTx(TokenTx):
def __str__(self):
if self.symmetric:
return '{} {} => {} {} {}'.format(
return '{}\t{} => {}\t{} {}'.format(
self.date_block_label,
self.sender_label,
self.recipient_label,

View File

@ -12,16 +12,13 @@ from cic_types.condiments import MetadataPointer
from cic_types.models.person import Person
from cic_types.ext.requests import make_request
from cic_types.processor import generate_metadata_pointer
from requests.exceptions import HTTPError
import requests.exceptions
import phonenumbers
# local imports
from clicada.encode import tx_normalize
from clicada.store.mem import MemDictStore
from clicada.error import (
ExpiredRecordError,
MetadataNotFoundError,
)
from clicada.error import ExpiredRecordError
logg = logging.getLogger(__name__)
@ -65,7 +62,7 @@ class Account(Person):
class FileUserStore:
def __init__(self, metadata_opener, chain_spec, label, store_base_path, ttl, encrypter=None):
def __init__(self, metadata_opener, chain_spec, label, store_base_path, ttl):
invalidate_before = datetime.datetime.now() - datetime.timedelta(seconds=ttl)
self.invalidate_before = int(invalidate_before.timestamp())
self.have_xattr = False
@ -81,8 +78,6 @@ class FileUserStore:
os.makedirs(self.store_path, exist_ok=True)
self.__validate_dir()
self.metadata_opener = metadata_opener
self.failed_entities = {}
self.encrypter = encrypter
def __validate_dir(self):
@ -91,10 +86,6 @@ class FileUserStore:
logg.debug('using existing file store {} for {}'.format(self.store_path, self.label))
def is_dud(self, address):
return bool(self.failed_entities.get(address))
def put(self, k, v, force=False):
have_file = False
p = os.path.join(self.store_path, k)
@ -109,14 +100,8 @@ class FileUserStore:
if have_file and not its_time and not force:
raise FileExistsError('user resolution already exists for {}'.format(k))
ve = v
f = None
if self.encrypter != None:
ve = self.encrypter.encrypt(k, ve.encode('utf-8'))
f = open(p, 'wb')
else:
f = open(p, 'w')
f.write(ve)
f = open(p, 'w')
f.write(v)
f.close()
logg.info('added user store {} record {} -> {}'.format(self.label, k, v))
@ -181,21 +166,12 @@ class FileUserStore:
self.__unstick(p)
self.check_expiry(p)
f = None
if self.encrypter != None:
f = open(p, 'rb')
else:
f = open(p, 'r')
v = f.read()
f = open(p, 'r')
r = f.read()
f.close()
if self.encrypter != None:
v = self.encrypter.decrypt(k, v)
logg.debug('>>>>>>>>>>>>< v decoded {}'.format(v))
v = v.decode('utf-8')
logg.debug('retrieved {} from {}'.format(k, p))
return v.strip()
return r.strip()
def by_phone(self, phone, update=False):
@ -222,44 +198,35 @@ class FileUserStore:
try:
r = getter.open(ptr)
user_address = json.loads(r)
except HTTPError as e:
except requests.exceptions.HTTPError as e:
logg.debug('no address found for phone {}: {}'.format(phone, e))
return None
self.put(phone_file, user_address, force=update)
return user_address
def metadata_to_person(self, v):
person = Account()
try:
person_data = person.deserialize(person_data=v)
except Exception as e:
person_data = v
return person_data
def by_address(self, address, update=False):
address = tx_normalize.wallet_address(address)
address = strip_0x(address)
#if self.failed_entities.get(address):
if self.is_dud(address):
logg.debug('already tried and failed {}, skipping'.format(address))
return address
add = tx_normalize.wallet_address(address)
ignore_expired = self.sticky(address)
if not update:
try:
v = self.get(address, ignore_expired=ignore_expired)
v = json.loads(v)
return self.metadata_to_person(v)
person = Account()
try:
person_data = person.deserialize(person_data=v)
except Exception as e:
person_data = v
return person_data
except FileNotFoundError:
pass
except ExpiredRecordError as e:
logg.info(e)
pass
address = strip_0x(address)
getter = self.metadata_opener
ptr = generate_metadata_pointer(bytes.fromhex(address), MetadataPointer.PERSON)
@ -268,10 +235,7 @@ class FileUserStore:
r = getter.open(ptr)
except Exception as e:
logg.debug('no metadata found for {}: {}'.format(address, e))
if not r:
self.failed_entities[address] = True
raise MetadataNotFoundError()
return address
data = json.loads(r)
person = Account()

View File

@ -1,10 +1,7 @@
usumbufu~=0.3.8
confini~=0.6.0
cic-eth-registry~=0.6.9
cic-types~=0.2.2
usumbufu~=0.3.3a3
confini~=0.5.1
cic-eth-registry~=0.6.1
cic-types~=0.2.1a5
phonenumbers==8.12.12
eth-erc20~=0.3.0
eth-erc20~=0.1.2
hexathon~=0.1.0
pycryptodome~=3.10.1
chainlib-eth~=0.1.0
chainlib~=0.1.0

View File

@ -1,6 +1,6 @@
[metadata]
name = clicada
version = 0.1.3
version = 0.0.2a2
description = CLI CRM tool for the cic-stack custodial wallet system
author = Louis Holbrook
author_email = dev@holbrook.no
@ -34,4 +34,3 @@ packages =
clicada.cli
clicada.tx
clicada.user
clicada.crypt