Commit 87b77dd4 authored by matograine's avatar matograine

remove old useless silkaj modules

parent 0eb7bc56
"""
Copyright 2016-2019 Maël Azimi <m.a@moul.re>
Silkaj is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
Silkaj is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with utils.silkaj. If not, see <https://www.gnu.org/licenses/>.
"""
name = "silkaj"
"""
Copyright 2016-2019 Maël Azimi <m.a@moul.re>
Silkaj is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
Silkaj is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with utils.silkaj. If not, see <https://www.gnu.org/licenses/>.
"""
#from click import command, argument, echo, confirm
from time import time
from tabulate import tabulate
from duniterpy.api.bma import wot, blockchain
from duniterpy.documents import BlockUID, block_uid, Identity, Certification
from utils.silkaj.auth import auth_method
from utils.silkaj.tools import convert_time, message_exit, coroutine
from utils.silkaj.network_tools import ClientInstance, HeadBlock
from utils.silkaj.blockchain_tools import BlockchainParams
from utils.silkaj.license import license_approval
from utils.silkaj.wot import is_member, get_informations_for_identity
#@command("cert", help="Send certification")
#@argument("id_to_certify")
@coroutine
async def send_certification(id_to_certify):
client = ClientInstance().client
id_to_certify = await get_informations_for_identity(id_to_certify)
main_id_to_certify = id_to_certify["uids"][0]
# Authentication
key = auth_method()
# Check whether current user is member
issuer_pubkey = key.pubkey
issuer = await is_member(issuer_pubkey)
if not issuer:
message_exit("Current identity is not member.")
if issuer_pubkey == id_to_certify["pubkey"]:
message_exit("You can’t certify yourself!")
# Check if the certification can be renewed
req = await client(wot.requirements, id_to_certify["pubkey"])
req = req["identities"][0]
for cert in req["certifications"]:
if cert["from"] == issuer_pubkey:
params = await BlockchainParams().params
# Ğ1: 0<–>2y - 2y + 2m
# ĞT: 0<–>4.8m - 4.8m + 12.5d
renewable = cert["expiresIn"] - params["sigValidity"] + params["sigReplay"]
if renewable > 0:
renewable_date = convert_time(time() + renewable, "date")
message_exit("Certification renewable the " + renewable_date)
# Check if the certification is already in the pending certifications
for pending_cert in req["pendingCerts"]:
if pending_cert["from"] == issuer_pubkey:
message_exit("Certification is currently been processed")
# Display license and ask for confirmation
head = await HeadBlock().head_block
currency = head["currency"]
license_approval(currency)
# Certification confirmation
await certification_confirmation(
issuer, issuer_pubkey, id_to_certify, main_id_to_certify
)
identity = Identity(
version=10,
currency=currency,
pubkey=id_to_certify["pubkey"],
uid=main_id_to_certify["uid"],
ts=block_uid(main_id_to_certify["meta"]["timestamp"]),
signature=main_id_to_certify["self"],
)
certification = Certification(
version=10,
currency=currency,
pubkey_from=issuer_pubkey,
identity=identity,
timestamp=BlockUID(head["number"], head["hash"]),
signature="",
)
# Sign document
certification.sign([key])
# Send certification document
response = await client(wot.certify, certification.signed_raw())
if response.status == 200:
print("Certification successfully sent.")
else:
print("Error while publishing certification: {0}".format(await response.text()))
await client.close()
async def certification_confirmation(
issuer, issuer_pubkey, id_to_certify, main_id_to_certify
):
cert = list()
cert.append(["Cert", "From", "–>", "To: Published: #block-hash date"])
client = ClientInstance().client
idty_timestamp = main_id_to_certify["meta"]["timestamp"]
block_uid_idty = block_uid(idty_timestamp)
block = await client(blockchain.block, block_uid_idty.number)
block_uid_date = (
": #" + idty_timestamp[:15] + "… " + convert_time(block["time"], "all")
)
cert.append(["ID", issuer["uid"], "–>", main_id_to_certify["uid"] + block_uid_date])
cert.append(["Pubkey", issuer_pubkey, "–>", id_to_certify["pubkey"]])
params = await BlockchainParams().params
cert_begins = convert_time(time(), "date")
cert_ends = convert_time(time() + params["sigValidity"], "date")
cert.append(["Valid", cert_begins, "—>", cert_ends])
echo(tabulate(cert, tablefmt="fancy_grid"))
confirm("Do you confirm sending this certification?", abort=True)
"""
Copyright 2016-2019 Maël Azimi <m.a@moul.re>
Silkaj is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
Silkaj is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with utils.silkaj. If not, see <https://www.gnu.org/licenses/>.
"""
# -*- coding: utf-8 -*-
#from click import group, help_option, version_option, option, pass_context
from utils.silkaj.tx import send_transaction
from utils.silkaj.tx_history import transaction_history
from utils.silkaj.money import cmd_amount
from utils.silkaj.cert import send_certification
from utils.silkaj.commands import (
currency_info,
difficulties,
network_info,
argos_info,
list_blocks,
)
from utils.silkaj.wot import received_sent_certifications, id_pubkey_correspondence
from utils.silkaj.auth import generate_auth_file
from utils.silkaj.license import license_command
from utils.silkaj.constants import SILKAJ_VERSION
#@group()
#@help_option("-h", "--help")
#@version_option(SILKAJ_VERSION, "-v", "--version")
#@option(
"--peer",
"-p",
help="Default endpoint will reach Ğ1 currency with `https://g1.duniter.org` endpoint.\
Custom endpoint can be specified with `-p` option followed by <domain>:<port>",
)
#@option(
"--gtest", "-gt", is_flag=True, help="ĞTest: `https://g1-test.duniter.org` endpoint"
)
#@option(
"--auth-scrypt",
"--scrypt",
is_flag=True,
help="Scrypt authentication: default method",
)
#@option("--nrp", help='Scrypt parameters: defaults N,r,p: "4096,16,1"')
#@option(
"--auth-file",
"-af",
is_flag=True,
help="Authentication file. Defaults to: './authfile'",
)
#@option(
"--file",
default="authfile",
show_default=True,
help="Path file specification with '--auth-file'",
)
#@option("--auth-seed", "--seed", is_flag=True, help="Seed hexadecimal authentication")
#@option("--auth-wif", "--wif", is_flag=True, help="WIF and EWIF authentication methods")
#@pass_context
def cli(ctx, peer, gtest, auth_scrypt, nrp, auth_file, file, auth_seed, auth_wif):
ctx.obj = dict()
ctx.ensure_object(dict)
ctx.obj["PEER"] = peer
ctx.obj["GTEST"] = gtest
ctx.obj["AUTH_SCRYPT"] = auth_scrypt
ctx.obj["AUTH_SCRYPT_PARAMS"] = nrp
ctx.obj["AUTH_FILE"] = auth_file
ctx.obj["AUTH_FILE_PATH"] = file
ctx.obj["AUTH_SEED"] = auth_seed
ctx.obj["AUTH_WIF"] = auth_wif
cli.add_command(argos_info)
cli.add_command(generate_auth_file)
cli.add_command(cmd_amount)
cli.add_command(list_blocks)
cli.add_command(send_certification)
cli.add_command(difficulties)
cli.add_command(transaction_history)
cli.add_command(id_pubkey_correspondence)
cli.add_command(currency_info)
cli.add_command(license_command)
cli.add_command(network_info)
cli.add_command(send_transaction)
cli.add_command(received_sent_certifications)
#@cli.command("about", help="Display program information")
def about():
print(
"\
\n #@#@#@#@#@#@#@#@#@#@#@#@#@\
\n #@#@#@ #@ #@#@#@\
\n #@#@#@ #@#@ #@#@#@#@#@#@ #@#@. Silkaj",
SILKAJ_VERSION,
"\
\n #@#@ #@#@#@ #@#@#@#@#@#@#@#@#@#@#@ #@#@,\
\n #@#@ #@#@#@ &#@#@#@#@#@#@#@#@#@#@#@#@#@ #@#@#@ Powerfull and lightweight command line client\
\n #@#@ #@#@#@ #@#@#@#@#@#@#@#@#@# #@#@#@#@ #@#@(\
\n #@#@ #@#@#@#@ #@#@#@#@#@#@#@#@#@ #@#@#@ #@#@ Built in Python for Duniter’s currencies: Ğ1 and Ğ1-Test\
\n #@#@ #@#@#@ #@#@#@#@#@#@#@#@ #@ #@#@#@ #@#@\
\n #@#@ #@#@#@ #@#@#@#@#@#@ #@#@#@#@ #@#@ #@#@ Authors: moul, tortue, jytou, cebash, cgeek\
\n #@#@ #@#@#@#@ #@#@#@ #@#@#@#@#@#@#@ #@#@ #@#@\
\n #@#@ #@#@#@#@* #@#@#@#@#@#@#@#@#@ #@# #@#@ Website: https://utils.silkaj.duniter.org\
\n #@#@ #@#@#@#@#@ #@#@#@#@#@#@#@#@#@#@ #@ ,#@#@\
\n #@#@ #@#@#@#@#@ #@#@#@#@#@#@#@#@#@#@ #@ ,#@#@ Repository: https://git.duniter.org/clients/python/silkaj\
\n #@#@#@ #@#@#@#@#@#@#@#@#@#@#@#@ #@ #@#@*\
\n #@#@#@ #@#@#@#@#@#@#@#@ #@ #@#@#@ License: GNU AGPLv3\
\n #@#@#@#@ #@#@ #@#@#@,\
\n #@#@#@#@#@#@#@#@#@#@#@#@#@#@#@\n",
)
This diff is collapsed.
"""
Copyright 2016-2019 Maël Azimi <m.a@moul.re>
Silkaj is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
Silkaj is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with utils.silkaj. If not, see <https://www.gnu.org/licenses/>.
"""
import webbrowser
#from click import command, echo_via_pager, confirm
def license_approval(currency):
if currency != "g1":
return
if confirm(
"You will be asked to approve Ğ1 license. Would you like to display it?"
):
display_license()
confirm("Do you approve Ğ1 license?", abort=True)
#@command("license", help="Display Ğ1 license")
def license_command():
display_license()
def display_license():
language = input("In which language would you like to display Ğ1 license [en/fr]? ")
if language == "en":
if not webbrowser.open("https://duniter.org/en/wiki/g1-license/"):
echo_via_pager(open("licence-G1/license/license_g1-en.rst").read())
else:
if not webbrowser.open("https://duniter.org/fr/wiki/licence-g1/"):
echo_via_pager(open("licence-G1/license/license_g1-fr-FR.rst").read())
"""
Copyright 2016-2019 Maël Azimi <m.a@moul.re>
Silkaj is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
Silkaj is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with utils.silkaj. If not, see <https://www.gnu.org/licenses/>.
"""
#from click import command, argument, option, echo_via_pager, get_terminal_size
from texttable import Texttable
from operator import itemgetter, neg, eq, ne
from time import time
from duniterpy.api.bma.tx import history
from duniterpy.documents.transaction import Transaction
from utils.silkaj.network_tools import ClientInstance
from utils.silkaj.tools import convert_time, coroutine
from utils.silkaj.crypto_tools import check_public_key
from utils.silkaj.wot import identity_of, identities_from_pubkeys
from utils.silkaj.money import get_amount_from_pubkey, amount_in_current_base, UDValue
from utils.silkaj.tools import CurrencySymbol
#@command("history", help="Display transaction history")
#@argument("pubkey")
#@option("--uids", "-u", is_flag=True, help="Display uids")
@coroutine
async def transaction_history(pubkey, uids):
if not check_public_key(pubkey, True):
return
client = ClientInstance().client
ud_value = await UDValue().ud_value
currency_symbol = await CurrencySymbol().symbol
header = await generate_header(pubkey, currency_symbol, ud_value)
received_txs, sent_txs = list(), list()
await get_transactions_history(client, pubkey, received_txs, sent_txs)
remove_duplicate_txs(received_txs, sent_txs)
txs_list = await generate_table(
received_txs, sent_txs, pubkey, ud_value, currency_symbol, uids
)
table = Texttable(max_width=get_terminal_size()[0])
table.add_rows(txs_list)
await client.close()
echo_via_pager(header + table.draw())
async def generate_header(pubkey, currency_symbol, ud_value):
try:
idty = await identity_of(pubkey)
except:
idty = dict([("uid", "")])
balance = await get_amount_from_pubkey(pubkey)
return "Transactions history from: {uid} {pubkey}\n\
Current balance: {balance} {currency}, {balance_ud} UD {currency} on the {date}\n\
".format(
uid=idty["uid"],
pubkey=pubkey,
currency=currency_symbol,
balance=balance[1] / 100,
balance_ud=round(balance[1] / ud_value, 2),
date=convert_time(time(), "all"),
)
async def get_transactions_history(client, pubkey, received_txs, sent_txs):
"""
Get transaction history
Store txs in Transaction object
"""
tx_history = await client(history, pubkey)
currency = tx_history["currency"]
for received in tx_history["history"]["received"]:
received_txs.append(Transaction.from_bma_history(currency, received))
for sent in tx_history["history"]["sent"]:
sent_txs.append(Transaction.from_bma_history(currency, sent))
def remove_duplicate_txs(received_txs, sent_txs):
"""
Remove duplicate transactions from history
Remove received tx which contains output back return
that we don’t want to displayed
A copy of received_txs is necessary to remove elements
"""
for received_tx in list(received_txs):
if received_tx in sent_txs:
received_txs.remove(received_tx)
async def generate_table(
received_txs, sent_txs, pubkey, ud_value, currency_symbol, uids
):
"""
Generate information in a list of lists for texttabe
Merge received and sent txs
Insert table header at the end not to disturb its generation
Sort txs temporarily
"""
received_txs_table, sent_txs_table = list(), list()
await parse_received_tx(received_txs_table, received_txs, pubkey, ud_value, uids)
await parse_sent_tx(sent_txs_table, sent_txs, pubkey, ud_value, uids)
txs_table = received_txs_table + sent_txs_table
table_titles = [
"Date",
"Issuers/Recipients",
"Amounts {}".format(currency_symbol),
"Amounts UD{}".format(currency_symbol),
"Comment",
]
txs_table.insert(0, table_titles)
txs_table.sort(key=itemgetter(0), reverse=True)
return txs_table
async def parse_received_tx(received_txs_table, received_txs, pubkey, ud_value, uids):
"""
Extract issuers’ pubkeys
Get identities from pubkeys
Convert time into human format
Assign identities
Get amounts and assign amounts and amounts_ud
Append comment
"""
issuers = list()
for received_tx in received_txs:
for issuer in received_tx.issuers:
issuers.append(issuer)
identities = await identities_from_pubkeys(issuers, uids)
for received_tx in received_txs:
tx_list = list()
tx_list.append(convert_time(received_tx.time, "all"))
tx_list.append(str())
for i, issuer in enumerate(received_tx.issuers):
tx_list[1] += prefix(None, None, i) + assign_idty_from_pubkey(
issuer, identities
)
amounts = tx_amount(received_tx, pubkey, received_func)[0]
tx_list.append(amounts / 100)
tx_list.append(amounts / ud_value)
tx_list.append(received_tx.comment)
received_txs_table.append(tx_list)
async def parse_sent_tx(sent_txs_table, sent_txs, pubkey, ud_value, uids):
"""
Extract recipients’ pubkeys from outputs
Get identities from pubkeys
Convert time into human format
Store "Total" and total amounts according to the number of outputs
If not output back return:
Assign amounts, amounts_ud, identities, and comment
"""
pubkeys = list()
for sent_tx in sent_txs:
outputs = tx_amount(sent_tx, pubkey, sent_func)[1]
for output in outputs:
if output_available(output.condition, ne, pubkey):
pubkeys.append(output.condition.left.pubkey)
identities = await identities_from_pubkeys(pubkeys, uids)
for sent_tx in sent_txs:
tx_list = list()
tx_list.append(convert_time(sent_tx.time, "all"))
total_amount, outputs = tx_amount(sent_tx, pubkey, sent_func)
if len(outputs) > 1:
tx_list.append("Total")
amounts = str(total_amount / 100)
amounts_ud = str(round(total_amount / ud_value, 2))
else:
tx_list.append(str())
amounts = str()
amounts_ud = str()
for i, output in enumerate(outputs):
if output_available(output.condition, ne, pubkey):
amounts += prefix(None, outputs, i) + str(
neg(amount_in_current_base(output)) / 100
)
amounts_ud += prefix(None, outputs, i) + str(
round(neg(amount_in_current_base(output)) / ud_value, 2)
)
tx_list[1] += prefix(tx_list[1], outputs, 0) + assign_idty_from_pubkey(
output.condition.left.pubkey, identities
)
tx_list.append(amounts)
tx_list.append(amounts_ud)
tx_list.append(sent_tx.comment)
sent_txs_table.append(tx_list)
def tx_amount(tx, pubkey, function):
"""
Determine transaction amount from output sources
"""
amount = 0
outputs = list()
for output in tx.outputs:
if output_available(output.condition, ne, pubkey):
outputs.append(output)
amount += function(output, pubkey)
return amount, outputs
def received_func(output, pubkey):
if output_available(output.condition, eq, pubkey):
return amount_in_current_base(output)
return 0
def sent_func(output, pubkey):
if output_available(output.condition, ne, pubkey):
return neg(amount_in_current_base(output))
return 0
def output_available(condition, comparison, value):
"""
Check if output source is available
Currently only handle simple SIG condition
XHX, CLTV, CSV should be handled when present in the blockchain
"""
if hasattr(condition.left, "pubkey"):
return comparison(condition.left.pubkey, value)
else:
return False
def assign_idty_from_pubkey(pubkey, identities):
idty = pubkey[:18]
for identity in identities:
if pubkey == identity["pubkey"]:
idty = identity["uid"] + " - " + pubkey[:13]
return idty
def prefix(tx_addresses, outputs, occurence):
"""
Pretty print with texttable
Break line when several values in a cell
Handle Total line in case of multi-output txs
"""
if tx_addresses == "Total":
return "\n"
if not outputs:
return ""
return "\n" if occurence + len(outputs) > 1 else ""
"""
Copyright 2016-2019 Maël Azimi <m.a@moul.re>
Silkaj is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
Silkaj is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with utils.silkaj. If not, see <https://www.gnu.org/licenses/>.
"""
##from click import command, argument
from time import time
from tabulate import tabulate
from collections import OrderedDict
from asyncio import sleep
from duniterpy.api.bma import wot, blockchain
from duniterpy.api.errors import DuniterError
from utils.silkaj.network_tools import ClientInstance
from utils.silkaj.crypto_tools import check_public_key
from utils.silkaj.tools import message_exit, convert_time, coroutine
from utils.silkaj.blockchain_tools import BlockchainParams
from utils.silkaj.constants import ASYNC_SLEEP
def get_sent_certifications(certs, time_first_block, params):
sent = list()
expire = list()
if certs["signed"]:
for cert in certs["signed"]:
sent.append(cert["uid"])
expire.append(
expiration_date_from_block_id(
cert["cert_time"]["block"], time_first_block, params
)
)
return sent, expire
#@command(
# "wot",
# help="Check received and sent certifications and consult the membership status of any given identity",
#)
#@argument("id")
@coroutine
async def received_sent_certifications(id):
"""
get searched id
get id of received and sent certifications
display in a table the result with the numbers
"""
client = ClientInstance().client
first_block = await client(blockchain.block, 1)
time_first_block = first_block["time"]
id_certs = await get_informations_for_identity(id)
certifications = OrderedDict()
params = await BlockchainParams().params
for certs in id_certs["uids"]:
if certs["uid"].lower() == id.lower():
pubkey = id_certs["pubkey"]
req = await client(wot.requirements, pubkey)
req = req["identities"][0]
certifications["received_expire"] = list()
certifications["received"] = list()
for cert in certs["others"]:
certifications["received_expire"].append(
expiration_date_from_block_id(
cert["meta"]["block_number"], time_first_block, params
)
)
certifications["received"].append(
cert_written_in_the_blockchain(req["certifications"], cert)
)
certifications["sent"], certifications[
"sent_expire"
] = get_sent_certifications(id_certs, time_first_block, params)
nbr_sent_certs = (
len(certifications["sent"]) if "sent" in certifications else 0
)
print(
"{0} ({1}) from block #{2}\nreceived {3} and sent {4}/{5} certifications:\n{6}\n{7}\n".format(
id,
pubkey[:5] + "…",
certs["meta"]["timestamp"][:15] + "…",
len(certifications["received"]),
nbr_sent_certs,
params["sigStock"],
tabulate(
certifications,
headers="keys",
tablefmt="orgtbl",
stralign="center",
),
"✔: Certifications written into the blockchain",
)
)
await membership_status(certifications, certs, pubkey, req)
await client.close()
def cert_written_in_the_blockchain(written_certs, certifieur):
for cert in written_certs:
if cert["from"] == certifieur["pubkey"]:
return certifieur["uids"][0] + " ✔"
return certifieur["uids"][0]
async def membership_status(certifications, certs, pubkey, req):
params = await BlockchainParams().params
if len(certifications["received"]) >= params["sigQty"]:
print(
"Membership expiration due to certification expirations: "
+ certifications["received_expire"][
len(certifications["received"]) - params["sigQty"]
]
)
member = await is_member(pubkey)
if member:
member = True
print("member:", member)
if req["revoked"]:
print(
"revoked:",
req["revoked"],
"\nrevoked on:",
convert_time(req["revoked_on"], "date") + "\n",
)
if not member and req["wasMember"]:
print("expired:", req["expired"], "\nwasMember:", req["wasMember"])
elif member:
print(