mirror of
https://github.com/yzernik/squeaknode.git
synced 2026-08-20 13:28:20 +02:00
bb15da7058
* Add squeak store class * Use squeak store class to save squeaks made from make method * Use squeak store class to get public squeak for server get * Use squeak store class for other squeak node methods
518 lines
19 KiB
Python
518 lines
19 KiB
Python
from __future__ import print_function
|
|
|
|
import time
|
|
|
|
import pytest
|
|
from squeak.core import CheckSqueak
|
|
from squeak.core.encryption import CEncryptedDecryptionKey
|
|
|
|
from proto import lnd_pb2 as ln
|
|
from proto import (
|
|
squeak_admin_pb2,
|
|
squeak_admin_pb2_grpc,
|
|
squeak_server_pb2,
|
|
squeak_server_pb2_grpc,
|
|
)
|
|
from tests.util import (
|
|
build_squeak_msg,
|
|
generate_challenge_proof,
|
|
generate_signing_key,
|
|
get_address,
|
|
get_challenge,
|
|
get_hash,
|
|
get_latest_block_info,
|
|
make_squeak,
|
|
squeak_from_msg,
|
|
string_to_hex,
|
|
)
|
|
|
|
|
|
def test_post_squeak(
|
|
server_stub, admin_stub, lightning_client, whitelisted_signing_key
|
|
):
|
|
# Post a squeak with a direct request to the server
|
|
block_height, block_hash = get_latest_block_info(lightning_client)
|
|
squeak = make_squeak(
|
|
whitelisted_signing_key, "hello from itest!", block_hash, block_height
|
|
)
|
|
squeak_hash = get_hash(squeak)
|
|
|
|
squeak_msg = build_squeak_msg(squeak)
|
|
post_response = server_stub.PostSqueak(
|
|
squeak_server_pb2.PostSqueakRequest(squeak=squeak_msg)
|
|
)
|
|
|
|
# Wait a few seconds for the squeak to be verified on the server.
|
|
time.sleep(5)
|
|
|
|
# Get the same squeak from the server
|
|
get_response = server_stub.GetSqueak(
|
|
squeak_server_pb2.GetSqueakRequest(hash=squeak_hash)
|
|
)
|
|
get_response_squeak = squeak_from_msg(get_response.squeak)
|
|
CheckSqueak(get_response_squeak, skipDecryptionCheck=True)
|
|
|
|
assert get_hash(get_response_squeak) == get_hash(squeak)
|
|
|
|
|
|
def test_post_squeak_not_whitelisted(
|
|
server_stub, admin_stub, lightning_client, nonwhitelisted_signing_key
|
|
):
|
|
# Post a squeak with a direct request to the server
|
|
block_height, block_hash = get_latest_block_info(lightning_client)
|
|
squeak = make_squeak(
|
|
nonwhitelisted_signing_key, "hello from itest!", block_hash, block_height
|
|
)
|
|
squeak_hash = get_hash(squeak)
|
|
|
|
squeak_msg = build_squeak_msg(squeak)
|
|
with pytest.raises(Exception):
|
|
server_stub.PostSqueak(squeak_server_pb2.PostSqueakRequest(squeak=squeak_msg))
|
|
|
|
|
|
def test_lookup_squeaks(server_stub, admin_stub, signing_profile_id, saved_squeak_hash):
|
|
# Get the squeak profile
|
|
get_squeak_profile_response = admin_stub.GetSqueakProfile(
|
|
squeak_admin_pb2.GetSqueakProfileRequest(profile_id=signing_profile_id,)
|
|
)
|
|
squeak_profile_address = get_squeak_profile_response.squeak_profile.address
|
|
|
|
# Lookup squeaks for the given signing profile
|
|
addresses = [squeak_profile_address]
|
|
lookup_response = server_stub.LookupSqueaks(
|
|
squeak_server_pb2.LookupSqueaksRequest(
|
|
addresses=addresses, min_block=0, max_block=99999999,
|
|
)
|
|
)
|
|
assert len(lookup_response.hashes) == 1
|
|
assert saved_squeak_hash in set(lookup_response.hashes)
|
|
|
|
|
|
def test_lookup_squeaks_empty_result(server_stub, admin_stub):
|
|
signing_key = generate_signing_key()
|
|
address = get_address(signing_key)
|
|
|
|
# Lookup squeaks for the given address
|
|
addresses = [address]
|
|
lookup_response = server_stub.LookupSqueaks(
|
|
squeak_server_pb2.LookupSqueaksRequest(
|
|
addresses=addresses, min_block=0, max_block=99999999,
|
|
)
|
|
)
|
|
assert len(lookup_response.hashes) == 0
|
|
|
|
|
|
def test_sell_squeak(server_stub, admin_stub, lightning_client, saved_squeak_hash):
|
|
# Check the server balance
|
|
get_balance_response = admin_stub.LndWalletBalance(ln.WalletBalanceRequest())
|
|
initial_server_balance = get_balance_response.total_balance
|
|
|
|
# Get the squeak from the server
|
|
get_response = server_stub.GetSqueak(
|
|
squeak_server_pb2.GetSqueakRequest(hash=saved_squeak_hash)
|
|
)
|
|
get_response_squeak = squeak_from_msg(get_response.squeak)
|
|
CheckSqueak(get_response_squeak, skipDecryptionCheck=True)
|
|
|
|
# Generate a challenge to verify the offer
|
|
expected_proof = generate_challenge_proof()
|
|
encryption_key = get_response_squeak.GetEncryptionKey()
|
|
challenge = get_challenge(encryption_key, expected_proof)
|
|
|
|
# Buy the squeak data key
|
|
buy_response = server_stub.BuySqueak(
|
|
squeak_server_pb2.BuySqueakRequest(hash=saved_squeak_hash, challenge=challenge,)
|
|
)
|
|
assert buy_response.offer.payment_request.startswith("ln")
|
|
|
|
# Check the offer challenge proof
|
|
assert buy_response.offer.proof == expected_proof
|
|
|
|
# Connect to the server lightning node
|
|
connect_peer_response = lightning_client.connect_peer(
|
|
buy_response.offer.pubkey, buy_response.offer.host
|
|
)
|
|
|
|
# List peers
|
|
list_peers_response = lightning_client.list_peers()
|
|
|
|
# Open channel to the server lightning node
|
|
pubkey_bytes = string_to_hex(buy_response.offer.pubkey)
|
|
open_channel_response = lightning_client.open_channel(pubkey_bytes, 1000000)
|
|
print("Opening channel...")
|
|
for update in open_channel_response:
|
|
if update.HasField("chan_open"):
|
|
channel_point = update.chan_open.channel_point
|
|
print("Channel now open: " + str(channel_point))
|
|
break
|
|
|
|
# List channels
|
|
list_channels_response = lightning_client.list_channels()
|
|
|
|
# Pay the invoice
|
|
payment = lightning_client.pay_invoice_sync(buy_response.offer.payment_request)
|
|
preimage = payment.payment_preimage
|
|
|
|
# Verify with the payment preimage and decryption key ciphertext
|
|
decryption_key_cipher_bytes = buy_response.offer.key_cipher
|
|
iv = buy_response.offer.iv
|
|
encrypted_decryption_key = CEncryptedDecryptionKey.from_bytes(
|
|
decryption_key_cipher_bytes
|
|
)
|
|
|
|
# Decrypt the decryption key
|
|
decryption_key = encrypted_decryption_key.get_decryption_key(preimage, iv)
|
|
serialized_decryption_key = decryption_key.get_bytes()
|
|
|
|
get_response_squeak.SetDecryptionKey(serialized_decryption_key)
|
|
CheckSqueak(get_response_squeak)
|
|
|
|
# Close the channel
|
|
time.sleep(2)
|
|
for update in lightning_client.close_channel(channel_point):
|
|
if update.HasField("chan_close"):
|
|
print("Channel closed.")
|
|
break
|
|
|
|
# Check the server balance
|
|
get_balance_response = admin_stub.LndWalletBalance(ln.WalletBalanceRequest())
|
|
final_server_balance = get_balance_response.total_balance
|
|
assert final_server_balance - initial_server_balance == 1000
|
|
|
|
|
|
def test_make_squeak(server_stub, admin_stub, signing_profile_id):
|
|
# Create a new squeak using the new profile
|
|
make_squeak_content = "Hello from the profile on the server!"
|
|
make_squeak_response = admin_stub.MakeSqueak(
|
|
squeak_admin_pb2.MakeSqueakRequest(
|
|
profile_id=signing_profile_id, content=make_squeak_content,
|
|
)
|
|
)
|
|
make_squeak_hash = make_squeak_response.squeak_hash
|
|
assert len(make_squeak_hash) == 32 * 2
|
|
|
|
# Get the new squeak from the server
|
|
get_squeak_response = server_stub.GetSqueak(
|
|
squeak_server_pb2.GetSqueakRequest(hash=bytes.fromhex(make_squeak_hash))
|
|
)
|
|
get_squeak_response_squeak = squeak_from_msg(get_squeak_response.squeak)
|
|
CheckSqueak(get_squeak_response_squeak, skipDecryptionCheck=True)
|
|
assert get_hash(get_squeak_response_squeak) == bytes.fromhex(make_squeak_hash)
|
|
|
|
# Get the squeak display item
|
|
get_squeak_display_response = admin_stub.GetSqueakDisplay(
|
|
squeak_admin_pb2.GetSqueakDisplayRequest(squeak_hash=make_squeak_hash,)
|
|
)
|
|
assert (
|
|
get_squeak_display_response.squeak_display_entry.content_str
|
|
== "Hello from the profile on the server!"
|
|
)
|
|
|
|
# Get the squeak profile
|
|
get_squeak_profile_response = admin_stub.GetSqueakProfile(
|
|
squeak_admin_pb2.GetSqueakProfileRequest(profile_id=signing_profile_id,)
|
|
)
|
|
squeak_profile_address = get_squeak_profile_response.squeak_profile.address
|
|
squeak_profile_name = get_squeak_profile_response.squeak_profile.profile_name
|
|
|
|
# Get all squeak displays for the known address
|
|
get_address_squeak_display_response = admin_stub.GetAddressSqueakDisplays(
|
|
squeak_admin_pb2.GetAddressSqueakDisplaysRequest(address=squeak_profile_address)
|
|
)
|
|
assert len(get_address_squeak_display_response.squeak_display_entries) == 1
|
|
for (
|
|
squeak_display_entry
|
|
) in get_address_squeak_display_response.squeak_display_entries:
|
|
assert squeak_display_entry.author_name == squeak_profile_name
|
|
assert squeak_display_entry.author_address == squeak_profile_address
|
|
|
|
|
|
def test_make_reply_squeak(
|
|
server_stub, admin_stub, saved_squeak_hash, signing_profile_id
|
|
):
|
|
# Make another squeak as a reply
|
|
reply_1_squeak_response = admin_stub.MakeSqueak(
|
|
squeak_admin_pb2.MakeSqueakRequest(
|
|
profile_id=signing_profile_id,
|
|
content="Reply #1",
|
|
replyto=saved_squeak_hash.hex(),
|
|
)
|
|
)
|
|
reply_1_squeak_hash = reply_1_squeak_response.squeak_hash
|
|
|
|
# Make a second squeak as a reply
|
|
reply_2_squeak_response = admin_stub.MakeSqueak(
|
|
squeak_admin_pb2.MakeSqueakRequest(
|
|
profile_id=signing_profile_id,
|
|
content="Reply #2",
|
|
replyto=reply_1_squeak_hash,
|
|
)
|
|
)
|
|
reply_2_squeak_hash = reply_2_squeak_response.squeak_hash
|
|
|
|
# Get the squeak and check that the reply field is correct
|
|
get_reply_squeak_display_response = admin_stub.GetSqueakDisplay(
|
|
squeak_admin_pb2.GetSqueakDisplayRequest(squeak_hash=reply_2_squeak_hash,)
|
|
)
|
|
assert (
|
|
get_reply_squeak_display_response.squeak_display_entry.squeak_hash
|
|
== reply_2_squeak_hash
|
|
)
|
|
assert (
|
|
get_reply_squeak_display_response.squeak_display_entry.reply_to
|
|
== reply_1_squeak_hash
|
|
)
|
|
|
|
# Get the ancestors of the latest reply squeak
|
|
get_ancestors_response = admin_stub.GetAncestorSqueakDisplays(
|
|
squeak_admin_pb2.GetAncestorSqueakDisplaysRequest(
|
|
squeak_hash=reply_2_squeak_hash,
|
|
)
|
|
)
|
|
assert len(get_ancestors_response.squeak_display_entries) == 3
|
|
|
|
|
|
def test_post_squeak_rate_limit(server_stub, admin_stub, whitelisted_signing_key):
|
|
# Make 10 squeak
|
|
for i in range(10):
|
|
try:
|
|
block_height, block_hash = get_latest_block_info(lightning_client)
|
|
squeak = make_squeak(
|
|
nonwhitelisted_signing_key, "hello from itest!", block_hash, block_height
|
|
)
|
|
squeak_msg = build_squeak_msg(squeak)
|
|
server_stub.PostSqueak(squeak_server_pb2.PostSqueakRequest(squeak=squeak_msg))
|
|
except Exception as e:
|
|
post_squeak_exception = e
|
|
assert post_squeak_exception is not None
|
|
|
|
|
|
def test_make_signing_profile(server_stub, admin_stub):
|
|
# Create a new signing profile
|
|
profile_name = "test_signing_profile_name"
|
|
create_signing_profile_response = admin_stub.CreateSigningProfile(
|
|
squeak_admin_pb2.CreateSigningProfileRequest(profile_name=profile_name,)
|
|
)
|
|
profile_id = create_signing_profile_response.profile_id
|
|
|
|
# Get the new squeak profile
|
|
get_squeak_profile_response = admin_stub.GetSqueakProfile(
|
|
squeak_admin_pb2.GetSqueakProfileRequest(profile_id=profile_id,)
|
|
)
|
|
assert get_squeak_profile_response.squeak_profile.profile_name == profile_name
|
|
squeak_profile_address = get_squeak_profile_response.squeak_profile.address
|
|
|
|
# Get all signing profiles
|
|
get_signing_profiles_response = admin_stub.GetSigningProfiles(
|
|
squeak_admin_pb2.GetSigningProfilesRequest()
|
|
)
|
|
signing_profile_names = [
|
|
profile.profile_name
|
|
for profile in get_signing_profiles_response.squeak_profiles
|
|
]
|
|
assert profile_name in signing_profile_names
|
|
|
|
# Get squeak profile by address
|
|
get_profile_by_address_response = admin_stub.GetSqueakProfileByAddress(
|
|
squeak_admin_pb2.GetSqueakProfileByAddressRequest(
|
|
address=squeak_profile_address
|
|
)
|
|
)
|
|
assert get_profile_by_address_response.squeak_profile.profile_name == profile_name
|
|
|
|
|
|
def test_make_contact_profile(server_stub, admin_stub):
|
|
# Create a new contact profile
|
|
contact_name = "test_contact_profile_name"
|
|
contact_signing_key = generate_signing_key()
|
|
contact_address = get_address(contact_signing_key)
|
|
create_contact_profile_response = admin_stub.CreateContactProfile(
|
|
squeak_admin_pb2.CreateContactProfileRequest(
|
|
profile_name=contact_name, address=contact_address,
|
|
)
|
|
)
|
|
contact_profile_id = create_contact_profile_response.profile_id
|
|
|
|
# Get all contact profiles
|
|
get_contact_profiles_response = admin_stub.GetContactProfiles(
|
|
squeak_admin_pb2.GetContactProfilesRequest()
|
|
)
|
|
contact_profile_names = [
|
|
profile.profile_name
|
|
for profile in get_contact_profiles_response.squeak_profiles
|
|
]
|
|
assert contact_name in contact_profile_names
|
|
|
|
|
|
def test_set_profile_whitelisted(server_stub, admin_stub, contact_profile_id):
|
|
# Get the existing profile
|
|
get_squeak_profile_response = admin_stub.GetSqueakProfile(
|
|
squeak_admin_pb2.GetSqueakProfileRequest(profile_id=contact_profile_id,)
|
|
)
|
|
assert get_squeak_profile_response.squeak_profile.whitelisted == False
|
|
|
|
# Set the profile to be whitelisted
|
|
admin_stub.SetSqueakProfileWhitelisted(
|
|
squeak_admin_pb2.SetSqueakProfileWhitelistedRequest(
|
|
profile_id=contact_profile_id, whitelisted=True,
|
|
)
|
|
)
|
|
|
|
# Get the squeak profile again
|
|
get_squeak_profile_response = admin_stub.GetSqueakProfile(
|
|
squeak_admin_pb2.GetSqueakProfileRequest(profile_id=contact_profile_id,)
|
|
)
|
|
assert get_squeak_profile_response.squeak_profile.whitelisted == True
|
|
|
|
|
|
def test_set_profile_following(server_stub, admin_stub, contact_profile_id):
|
|
# Get the existing profile
|
|
get_squeak_profile_response = admin_stub.GetSqueakProfile(
|
|
squeak_admin_pb2.GetSqueakProfileRequest(profile_id=contact_profile_id,)
|
|
)
|
|
assert get_squeak_profile_response.squeak_profile.following == False
|
|
|
|
# Set the profile to be following
|
|
admin_stub.SetSqueakProfileFollowing(
|
|
squeak_admin_pb2.SetSqueakProfileFollowingRequest(
|
|
profile_id=contact_profile_id, following=True,
|
|
)
|
|
)
|
|
|
|
# Get the squeak profile again
|
|
get_squeak_profile_response = admin_stub.GetSqueakProfile(
|
|
squeak_admin_pb2.GetSqueakProfileRequest(profile_id=contact_profile_id,)
|
|
)
|
|
assert get_squeak_profile_response.squeak_profile.following == True
|
|
|
|
|
|
def test_set_profile_sharing(server_stub, admin_stub, contact_profile_id):
|
|
# Get the existing profile
|
|
get_squeak_profile_response = admin_stub.GetSqueakProfile(
|
|
squeak_admin_pb2.GetSqueakProfileRequest(profile_id=contact_profile_id,)
|
|
)
|
|
assert get_squeak_profile_response.squeak_profile.sharing == False
|
|
|
|
# Set the profile to be sharing
|
|
admin_stub.SetSqueakProfileSharing(
|
|
squeak_admin_pb2.SetSqueakProfileSharingRequest(
|
|
profile_id=contact_profile_id, sharing=True,
|
|
)
|
|
)
|
|
|
|
# Get the squeak profile again
|
|
get_squeak_profile_response = admin_stub.GetSqueakProfile(
|
|
squeak_admin_pb2.GetSqueakProfileRequest(profile_id=contact_profile_id,)
|
|
)
|
|
assert get_squeak_profile_response.squeak_profile.sharing == True
|
|
|
|
|
|
def test_get_following_squeaks(
|
|
server_stub, admin_stub, saved_squeak_hash, signing_profile_id
|
|
):
|
|
# Set the profile to be following
|
|
admin_stub.SetSqueakProfileFollowing(
|
|
squeak_admin_pb2.SetSqueakProfileFollowingRequest(
|
|
profile_id=signing_profile_id, following=True,
|
|
)
|
|
)
|
|
|
|
# Get all squeak displays for the known address
|
|
get_followed_squeak_display_response = admin_stub.GetFollowedSqueakDisplays(
|
|
squeak_admin_pb2.GetFollowedSqueakDisplaysRequest()
|
|
)
|
|
assert len(get_followed_squeak_display_response.squeak_display_entries) == 1
|
|
for (
|
|
squeak_display_entry
|
|
) in get_followed_squeak_display_response.squeak_display_entries:
|
|
# TODO: check the profile id of the squeak display entry
|
|
# assert squeak_display_entry.profile_id == signing_profile_id
|
|
pass
|
|
|
|
|
|
def test_delete_squeak(server_stub, admin_stub, saved_squeak_hash):
|
|
# Delete the squeak
|
|
admin_stub.DeleteSqueak(
|
|
squeak_admin_pb2.DeleteSqueakRequest(squeak_hash=saved_squeak_hash.hex())
|
|
)
|
|
|
|
# Try to get the squeak and fail
|
|
with pytest.raises(Exception):
|
|
get_response = server_stub.GetSqueak(
|
|
squeak_server_pb2.GetSqueakRequest(hash=saved_squeak_hash)
|
|
)
|
|
|
|
|
|
def test_create_subscription(server_stub, admin_stub):
|
|
# Add a new subscription
|
|
create_subscription_response = admin_stub.CreateSubscription(
|
|
squeak_admin_pb2.CreateSubscriptionRequest(host="fake_host", port=1234,)
|
|
)
|
|
subscription_id = create_subscription_response.subscription_id
|
|
|
|
# Get the new subscription
|
|
get_subscription_response = admin_stub.GetSubscription(
|
|
squeak_admin_pb2.GetSubscriptionRequest(
|
|
subscription_id=subscription_id,
|
|
)
|
|
)
|
|
assert get_subscription_response.squeak_subscription.host == "fake_host"
|
|
assert get_subscription_response.squeak_subscription.port == 1234
|
|
|
|
# Get all subscriptions
|
|
get_subscriptions_response = admin_stub.GetSubscriptions(
|
|
squeak_admin_pb2.GetSubscriptionsRequest()
|
|
)
|
|
subscription_hosts = [
|
|
squeak_subscription.host
|
|
for squeak_subscription in get_subscriptions_response.squeak_subscriptions
|
|
]
|
|
assert "fake_host" in subscription_hosts
|
|
|
|
def test_set_subscription_subscribed(server_stub, admin_stub, subscription_id):
|
|
# Get the subscription
|
|
get_subscription_response = admin_stub.GetSubscription(
|
|
squeak_admin_pb2.GetSubscriptionRequest(
|
|
subscription_id=subscription_id,
|
|
)
|
|
)
|
|
assert get_subscription_response.squeak_subscription.subscribed == False
|
|
|
|
# Set the subscription to be subscribed
|
|
admin_stub.SetSubscriptionSubscribed(
|
|
squeak_admin_pb2.SetSubscriptionSubscribedRequest(
|
|
subscription_id=subscription_id, subscribed=True,
|
|
)
|
|
)
|
|
|
|
# Get the subscription again
|
|
get_subscription_response = admin_stub.GetSubscription(
|
|
squeak_admin_pb2.GetSubscriptionRequest(
|
|
subscription_id=subscription_id,
|
|
)
|
|
)
|
|
assert get_subscription_response.squeak_subscription.subscribed == True
|
|
|
|
def test_set_subscription_publishing(server_stub, admin_stub, subscription_id):
|
|
# Get the subscription
|
|
get_subscription_response = admin_stub.GetSubscription(
|
|
squeak_admin_pb2.GetSubscriptionRequest(
|
|
subscription_id=subscription_id,
|
|
)
|
|
)
|
|
assert get_subscription_response.squeak_subscription.publishing == False
|
|
|
|
# Set the subscription to be publishing
|
|
admin_stub.SetSubscriptionPublishing(
|
|
squeak_admin_pb2.SetSubscriptionPublishingRequest(
|
|
subscription_id=subscription_id, publishing=True,
|
|
)
|
|
)
|
|
|
|
# Get the subscription again
|
|
get_subscription_response = admin_stub.GetSubscription(
|
|
squeak_admin_pb2.GetSubscriptionRequest(
|
|
subscription_id=subscription_id,
|
|
)
|
|
)
|
|
assert get_subscription_response.squeak_subscription.publishing == True
|