mirror of
https://github.com/aljazceru/nutshell.git
synced 2026-02-10 19:24:22 +01:00
Type annotations for bdhke (#140)
* annotate dhke * raise for errors * remove backwards compatibility * remove old code * remove test code
This commit is contained in:
@@ -35,7 +35,7 @@ import hashlib
|
||||
from secp256k1 import PrivateKey, PublicKey
|
||||
|
||||
|
||||
def hash_to_curve(message: bytes):
|
||||
def hash_to_curve(message: bytes) -> PublicKey:
|
||||
"""Generates a point from the message hash and checks if the point lies on the curve.
|
||||
If it does not, it tries computing a new point from the hash."""
|
||||
point = None
|
||||
@@ -49,28 +49,30 @@ def hash_to_curve(message: bytes):
|
||||
return point
|
||||
|
||||
|
||||
def step1_alice(secret_msg: str, blinding_factor: bytes = None):
|
||||
Y = hash_to_curve(secret_msg.encode("utf-8"))
|
||||
def step1_alice(
|
||||
secret_msg: str, blinding_factor: bytes = None
|
||||
) -> tuple[PublicKey, PrivateKey]:
|
||||
Y: PublicKey = hash_to_curve(secret_msg.encode("utf-8"))
|
||||
if blinding_factor:
|
||||
r = PrivateKey(privkey=blinding_factor, raw=True)
|
||||
else:
|
||||
r = PrivateKey()
|
||||
B_ = Y + r.pubkey
|
||||
B_: PublicKey = Y + r.pubkey
|
||||
return B_, r
|
||||
|
||||
|
||||
def step2_bob(B_, a):
|
||||
C_ = B_.mult(a)
|
||||
def step2_bob(B_: PublicKey, a: PrivateKey) -> PublicKey:
|
||||
C_: PublicKey = B_.mult(a)
|
||||
return C_
|
||||
|
||||
|
||||
def step3_alice(C_, r, A):
|
||||
C = C_ - A.mult(r)
|
||||
def step3_alice(C_: PublicKey, r: PrivateKey, A: PublicKey) -> PublicKey:
|
||||
C: PublicKey = C_ - A.mult(r)
|
||||
return C
|
||||
|
||||
|
||||
def verify(a, C, secret_msg):
|
||||
Y = hash_to_curve(secret_msg.encode("utf-8"))
|
||||
def verify(a: PrivateKey, C: PublicKey, secret_msg: str) -> bool:
|
||||
Y: PublicKey = hash_to_curve(secret_msg.encode("utf-8"))
|
||||
return C == Y.mult(a)
|
||||
|
||||
|
||||
|
||||
@@ -30,6 +30,7 @@ class LNbitsWallet(Wallet):
|
||||
async def status(self) -> StatusResponse:
|
||||
try:
|
||||
r = self.s.get(url=f"{self.endpoint}/api/v1/wallet", timeout=15)
|
||||
r.raise_for_status()
|
||||
except Exception as exc:
|
||||
return StatusResponse(
|
||||
f"Failed to connect to {self.endpoint} due to: {exc}", 0
|
||||
@@ -61,6 +62,7 @@ class LNbitsWallet(Wallet):
|
||||
data["memo"] = memo or ""
|
||||
try:
|
||||
r = self.s.post(url=f"{self.endpoint}/api/v1/payments", json=data)
|
||||
r.raise_for_status()
|
||||
except:
|
||||
return InvoiceResponse(False, None, None, r.json()["detail"])
|
||||
ok, checking_id, payment_request, error_message = (
|
||||
@@ -82,6 +84,7 @@ class LNbitsWallet(Wallet):
|
||||
json={"out": True, "bolt11": bolt11},
|
||||
timeout=None,
|
||||
)
|
||||
r.raise_for_status()
|
||||
except:
|
||||
error_message = r.json()["detail"]
|
||||
return PaymentResponse(None, None, None, None, error_message)
|
||||
@@ -112,6 +115,7 @@ class LNbitsWallet(Wallet):
|
||||
url=f"{self.endpoint}/api/v1/payments/{checking_id}",
|
||||
headers=self.key,
|
||||
)
|
||||
r.raise_for_status()
|
||||
except:
|
||||
return PaymentStatus(None)
|
||||
if r.json().get("detail"):
|
||||
@@ -123,6 +127,7 @@ class LNbitsWallet(Wallet):
|
||||
r = self.s.get(
|
||||
url=f"{self.endpoint}/api/v1/payments/{checking_id}", headers=self.key
|
||||
)
|
||||
r.raise_for_status()
|
||||
except:
|
||||
return PaymentStatus(None)
|
||||
data = r.json()
|
||||
|
||||
@@ -129,15 +129,6 @@ class Ledger:
|
||||
]
|
||||
|
||||
C = PublicKey(bytes.fromhex(proof.C), raw=True)
|
||||
|
||||
# backwards compatibility with old hash_to_curve < 0.4.0
|
||||
try:
|
||||
ret = legacy.verify_pre_0_3_3(private_key_amount, C, proof.secret)
|
||||
if ret:
|
||||
return ret
|
||||
except:
|
||||
pass
|
||||
|
||||
return b_dhke.verify(private_key_amount, C, proof.secret)
|
||||
|
||||
def _verify_script(self, idx: int, proof: Proof):
|
||||
|
||||
Reference in New Issue
Block a user