262 lines
12 KiB
Python
262 lines
12 KiB
Python
import json
|
|
import os
|
|
import requests
|
|
import sys
|
|
import unittest
|
|
import uuid
|
|
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from cryptography.hazmat.primitives import serialization
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', 'wikideck')))
|
|
|
|
from Mine.Block import Block
|
|
from Mine.Transaction import Transaction
|
|
|
|
WIKIDECK_URL = os.getenv('WIKIDECK_URL', 'http://localhost:8080/')
|
|
|
|
class TestMine(unittest.TestCase):
|
|
def test_mine_block_send_card_mine_another_block(self):
|
|
# Create two users
|
|
privateKeyA = rsa.generate_private_key(
|
|
public_exponent=65537,
|
|
key_size=4096
|
|
)
|
|
privateKeyB = rsa.generate_private_key(
|
|
public_exponent=65537,
|
|
key_size=4096
|
|
)
|
|
# Mine a block with privateKeyA
|
|
response = requests.get(f"{WIKIDECK_URL}/blocks")
|
|
self.assertEqual(response.status_code, 200)
|
|
newBlockA = Block(
|
|
data = response.json()
|
|
)
|
|
newBlockA.mine(privateKeyA)
|
|
response = requests.post(f"{WIKIDECK_URL}/blocks", json=newBlockA.as_dict())
|
|
self.assertEqual(response.status_code, 200)
|
|
# Send card from newBlockA to privateKeyB
|
|
newTransaction = Transaction(
|
|
cardId = newBlockA.card.cardId,
|
|
sender = privateKeyA.public_key(),
|
|
receiver = privateKeyB.public_key(),
|
|
authorPrivateKey = privateKeyA
|
|
)
|
|
response = requests.post(f"{WIKIDECK_URL}/transactions", json=newTransaction.as_dict())
|
|
self.assertEqual(response.status_code, 200)
|
|
# Mine a block with privateKeyB
|
|
response = requests.get(f"{WIKIDECK_URL}/blocks")
|
|
self.assertEqual(response.status_code, 200)
|
|
newBlockB = Block(
|
|
data = response.json()
|
|
)
|
|
newBlockB.mine(privateKeyB)
|
|
self.assertEqual(newBlockA.blockHash.hexdigest(), newBlockB.previousHash)
|
|
self.assertTrue(newBlockA.timestamp < newBlockB.timestamp)
|
|
self.assertTrue(newBlockB.height == newBlockA.height + 1)
|
|
response = requests.post(f"{WIKIDECK_URL}/blocks", json=newBlockB.as_dict())
|
|
self.assertEqual(response.status_code, 200)
|
|
# Validate decks
|
|
response = requests.get(f"{WIKIDECK_URL}/cards", params={
|
|
'publicKey': privateKeyA.public_key().public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
).decode('utf-8')
|
|
})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.json()), 0)
|
|
response = requests.get(f"{WIKIDECK_URL}/cards", params={
|
|
'publicKey': privateKeyB.public_key().public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
).decode('utf-8')
|
|
})
|
|
self.assertEqual(response.json()[0]['cardId'], str(newBlockA.card.cardId))
|
|
self.assertEqual(response.json()[0]['pageId'], newBlockA.card.pageId)
|
|
self.assertEqual(response.json()[1]['cardId'], str(newBlockB.card.cardId))
|
|
self.assertEqual(response.json()[1]['pageId'], newBlockB.card.pageId)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.json()), 2)
|
|
|
|
def test_invalid_block_modified_transaction(self):
|
|
# Create two users
|
|
privateKeyA = rsa.generate_private_key(
|
|
public_exponent=65537,
|
|
key_size=4096
|
|
)
|
|
privateKeyB = rsa.generate_private_key(
|
|
public_exponent=65537,
|
|
key_size=4096
|
|
)
|
|
# Mine a block with privateKeyA
|
|
response = requests.get(f"{WIKIDECK_URL}/blocks")
|
|
self.assertEqual(response.status_code, 200)
|
|
newBlockA = Block(
|
|
data = response.json()
|
|
)
|
|
newBlockA.mine(privateKeyA)
|
|
response = requests.post(f"{WIKIDECK_URL}/blocks", json=newBlockA.as_dict())
|
|
self.assertEqual(response.status_code, 200)
|
|
# Send card from newBlockA to privateKeyB
|
|
newTransaction = Transaction(
|
|
cardId = newBlockA.card.cardId,
|
|
sender = privateKeyA.public_key(),
|
|
receiver = privateKeyB.public_key(),
|
|
authorPrivateKey = privateKeyA
|
|
)
|
|
response = requests.post(f"{WIKIDECK_URL}/transactions", json=newTransaction.as_dict())
|
|
self.assertEqual(response.status_code, 200)
|
|
# Mine a block with privateKeyB
|
|
response = requests.get(f"{WIKIDECK_URL}/blocks")
|
|
self.assertEqual(response.status_code, 200)
|
|
newBlockB = Block(
|
|
data = response.json()
|
|
)
|
|
# Modify transactions
|
|
newBlockB.transactions[0].sender = privateKeyB.public_key()
|
|
newBlockB.transactions[0].receiver = privateKeyA.public_key()
|
|
newBlockB.mine(privateKeyB)
|
|
self.assertEqual(newBlockA.blockHash.hexdigest(), newBlockB.previousHash)
|
|
self.assertTrue(newBlockA.timestamp < newBlockB.timestamp)
|
|
self.assertTrue(newBlockB.height == newBlockA.height + 1)
|
|
response = requests.post(f"{WIKIDECK_URL}/blocks", json=newBlockB.as_dict())
|
|
self.assertEqual(response.status_code, 400)
|
|
# Validate decks
|
|
response = requests.get(f"{WIKIDECK_URL}/cards", params={
|
|
'publicKey': privateKeyA.public_key().public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
).decode('utf-8')
|
|
})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.json()), 1)
|
|
self.assertEqual(response.json()[0]['cardId'], str(newBlockA.card.cardId))
|
|
self.assertEqual(response.json()[0]['pageId'], newBlockA.card.pageId)
|
|
response = requests.get(f"{WIKIDECK_URL}/cards", params={
|
|
'publicKey': privateKeyB.public_key().public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
).decode('utf-8')
|
|
})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.json()), 0)
|
|
# Mine a block with privateKeyB
|
|
response = requests.get(f"{WIKIDECK_URL}/blocks")
|
|
self.assertEqual(response.status_code, 200)
|
|
newBlockB = Block(
|
|
data = response.json()
|
|
)
|
|
newBlockB.mine(privateKeyB)
|
|
self.assertEqual(newBlockA.blockHash.hexdigest(), newBlockB.previousHash)
|
|
self.assertTrue(newBlockA.timestamp < newBlockB.timestamp)
|
|
self.assertTrue(newBlockB.height == newBlockA.height + 1)
|
|
response = requests.post(f"{WIKIDECK_URL}/blocks", json=newBlockB.as_dict())
|
|
self.assertEqual(response.status_code, 200)
|
|
# Validate decks
|
|
response = requests.get(f"{WIKIDECK_URL}/cards", params={
|
|
'publicKey': privateKeyA.public_key().public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
).decode('utf-8')
|
|
})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.json()), 0)
|
|
response = requests.get(f"{WIKIDECK_URL}/cards", params={
|
|
'publicKey': privateKeyB.public_key().public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
).decode('utf-8')
|
|
})
|
|
self.assertEqual(response.json()[0]['cardId'], str(newBlockA.card.cardId))
|
|
self.assertEqual(response.json()[0]['pageId'], newBlockA.card.pageId)
|
|
self.assertEqual(response.json()[1]['cardId'], str(newBlockB.card.cardId))
|
|
self.assertEqual(response.json()[1]['pageId'], newBlockB.card.pageId)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.json()), 2)
|
|
|
|
def test_invalid_transaction(self):
|
|
# Create three users
|
|
privateKeyA = rsa.generate_private_key(
|
|
public_exponent=65537,
|
|
key_size=4096
|
|
)
|
|
privateKeyB = rsa.generate_private_key(
|
|
public_exponent=65537,
|
|
key_size=4096
|
|
)
|
|
privateKeyC = rsa.generate_private_key(
|
|
public_exponent=65537,
|
|
key_size=4096
|
|
)
|
|
# Mine a block with privateKeyA
|
|
response = requests.get(f"{WIKIDECK_URL}/blocks")
|
|
self.assertEqual(response.status_code, 200)
|
|
newBlockA = Block(
|
|
data = response.json()
|
|
)
|
|
newBlockA.mine(privateKeyA)
|
|
response = requests.post(f"{WIKIDECK_URL}/blocks", json=newBlockA.as_dict())
|
|
self.assertEqual(response.status_code, 200)
|
|
# Send card from newBlockA to privateKeyB
|
|
newTransaction = Transaction(
|
|
cardId = newBlockA.card.cardId,
|
|
sender = privateKeyA.public_key(),
|
|
receiver = privateKeyB.public_key(),
|
|
authorPrivateKey = privateKeyA
|
|
)
|
|
# Invalidate transaction
|
|
newTransaction.cardId = uuid.uuid4()
|
|
response = requests.post(f"{WIKIDECK_URL}/transactions", json=newTransaction.as_dict())
|
|
self.assertEqual(response.status_code, 400)
|
|
newTransaction.cardId = newBlockA.card.cardId
|
|
newTransaction.sender = privateKeyC.public_key()
|
|
response = requests.post(f"{WIKIDECK_URL}/transactions", json=newTransaction.as_dict())
|
|
self.assertEqual(response.status_code, 400)
|
|
newTransaction.sender = privateKeyA.public_key()
|
|
newTransaction.receiver = privateKeyC.public_key()
|
|
response = requests.post(f"{WIKIDECK_URL}/transactions", json=newTransaction.as_dict())
|
|
self.assertEqual(response.status_code, 400)
|
|
newTransaction.receiver = privateKeyB.public_key()
|
|
newTransaction.sign(privateKeyC)
|
|
response = requests.post(f"{WIKIDECK_URL}/transactions", json=newTransaction.as_dict())
|
|
self.assertEqual(response.status_code, 400)
|
|
newTransaction.sign(privateKeyA)
|
|
response = requests.post(f"{WIKIDECK_URL}/transactions", json=newTransaction.as_dict())
|
|
self.assertEqual(response.status_code, 200)
|
|
# Mine a block with privateKeyB
|
|
response = requests.get(f"{WIKIDECK_URL}/blocks")
|
|
self.assertEqual(response.status_code, 200)
|
|
newBlockB = Block(
|
|
data = response.json()
|
|
)
|
|
newBlockB.mine(privateKeyB)
|
|
self.assertEqual(newBlockA.blockHash.hexdigest(), newBlockB.previousHash)
|
|
self.assertTrue(newBlockA.timestamp < newBlockB.timestamp)
|
|
self.assertTrue(newBlockB.height == newBlockA.height + 1)
|
|
response = requests.post(f"{WIKIDECK_URL}/blocks", json=newBlockB.as_dict())
|
|
self.assertEqual(response.status_code, 200)
|
|
# Validate decks
|
|
response = requests.get(f"{WIKIDECK_URL}/cards", params={
|
|
'publicKey': privateKeyA.public_key().public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
).decode('utf-8')
|
|
})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.json()), 0)
|
|
response = requests.get(f"{WIKIDECK_URL}/cards", params={
|
|
'publicKey': privateKeyB.public_key().public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
).decode('utf-8')
|
|
})
|
|
self.assertEqual(response.json()[0]['cardId'], str(newBlockA.card.cardId))
|
|
self.assertEqual(response.json()[0]['pageId'], newBlockA.card.pageId)
|
|
self.assertEqual(response.json()[1]['cardId'], str(newBlockB.card.cardId))
|
|
self.assertEqual(response.json()[1]['pageId'], newBlockB.card.pageId)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.json()), 2)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|