diff --git a/CHANGELOG.md b/CHANGELOG.md index 141fb297..6f86a69d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ ml). ## [1.0.5-beta] - [addend] #124 adding endpoint for extract the internal stats of use +- [addend] #122 system for verify all documents that it's produced from devicehub ## [1.0.4-beta] - [addend] #95 adding endpoint for check the hash of one report diff --git a/ereuse_devicehub/__init__.py b/ereuse_devicehub/__init__.py index a20c909b..53f0b176 100644 --- a/ereuse_devicehub/__init__.py +++ b/ereuse_devicehub/__init__.py @@ -1 +1 @@ -__version__ = "1.0.4-beta" +__version__ = "1.0.5-beta" diff --git a/ereuse_devicehub/resources/documents/documents.py b/ereuse_devicehub/resources/documents/documents.py index 692e6992..cc47fd30 100644 --- a/ereuse_devicehub/resources/documents/documents.py +++ b/ereuse_devicehub/resources/documents/documents.py @@ -27,7 +27,7 @@ from ereuse_devicehub.resources.documents.device_row import (DeviceRow, StockRow InternalStatsRow) from ereuse_devicehub.resources.lot import LotView from ereuse_devicehub.resources.lot.models import Lot -from ereuse_devicehub.resources.hash_reports import insert_hash, ReportHash +from ereuse_devicehub.resources.hash_reports import insert_hash, ReportHash, verify_hash class Format(enum.Enum): @@ -80,6 +80,7 @@ class DocumentView(DeviceView): res = flask_weasyprint.render_pdf( flask_weasyprint.HTML(string=template), download_filename='{}.pdf'.format(type) ) + insert_hash(res.data) else: res = flask.make_response(template) return res @@ -186,7 +187,9 @@ class LotsDocumentView(LotView): cw.writerow(l.keys()) first = False cw.writerow(l.values()) - output = make_response(data.getvalue()) + bfile = data.getvalue().encode('utf-8') + output = make_response(bfile) + insert_hash(bfile) output.headers['Content-Disposition'] = 'attachment; filename=lots-info.csv' output.headers['Content-type'] = 'text/csv' return output @@ -223,7 +226,9 @@ class StockDocumentView(DeviceView): cw.writerow(d.keys()) first = False cw.writerow(d.values()) - output = make_response(data.getvalue()) + bfile = data.getvalue().encode('utf-8') + output = make_response(bfile) + insert_hash(bfile) output.headers['Content-Disposition'] = 'attachment; filename=devices-stock.csv' output.headers['Content-type'] = 'text/csv' return output @@ -247,12 +252,32 @@ class StampsView(View): This view render one public ans static page for see the links for to do the check of one csv file """ - def get(self): + def get_url_path(self): url = urlutils.URL(request.url) url.normalize() url.path_parts = url.path_parts[:-2] + ['check', ''] - url_path = url.to_text() - return flask.render_template('documents/stamp.html', rq_url=url_path) + return url.to_text() + + def get(self): + result = ('', '') + return flask.render_template('documents/stamp.html', rq_url=self.get_url_path(), + result=result) + + def post(self): + result = ('', '') + if 'docUpload' in request.files: + file_check = request.files['docUpload'] + bad = 'There are no coincidences. The attached file data does not come \ + from our backend or it has been subsequently modified.' + ok = '100% coincidence. The attached file contains data 100% existing in \ + to our backend' + result = ('Bad', bad) + if file_check.mimetype in ['text/csv', 'application/pdf']: + if verify_hash(file_check): + result = ('Ok', ok) + + return flask.render_template('documents/stamp.html', rq_url=self.get_url_path(), + result=result) class InternalStatsView(DeviceView): @@ -345,7 +370,7 @@ class DocumentDef(Resource): self.add_url_rule('/check/', defaults={}, view_func=check_view, methods=get) stamps_view = StampsView.as_view('StampsView', definition=self, auth=app.auth) - self.add_url_rule('/stamps/', defaults={}, view_func=stamps_view, methods=get) + self.add_url_rule('/stamps/', defaults={}, view_func=stamps_view, methods={'GET', 'POST'}) internalstats_view = InternalStatsView.as_view( 'InternalStatsView', definition=self, auth=app.auth) diff --git a/ereuse_devicehub/resources/documents/templates/documents/stamp.html b/ereuse_devicehub/resources/documents/templates/documents/stamp.html index c204c2ff..d07efc49 100644 --- a/ereuse_devicehub/resources/documents/templates/documents/stamp.html +++ b/ereuse_devicehub/resources/documents/templates/documents/stamp.html @@ -38,6 +38,17 @@
+
+ {% if result.0 == 'Ok' %} + + {% elif result.0 == 'Bad' %} + + {% endif %} +
Add one new check in your csv
@@ -45,6 +56,19 @@ Verify a CSV file in here.
+ +
+
+ If you want us to verify a document issued by us, upload it using the following form +
+
+
+ +
+ +
+
+
diff --git a/ereuse_devicehub/resources/hash_reports.py b/ereuse_devicehub/resources/hash_reports.py index f84e4b18..f44a79be 100644 --- a/ereuse_devicehub/resources/hash_reports.py +++ b/ereuse_devicehub/resources/hash_reports.py @@ -32,3 +32,8 @@ def insert_hash(bfile): db.session.add(db_hash) db.session.commit() db.session.flush() + + +def verify_hash(bfile): + hash3 = hashlib.sha3_256(bfile.read()).hexdigest() + return ReportHash.query.filter(ReportHash.hash3 == hash3).count() diff --git a/tests/test_documents.py b/tests/test_documents.py index 5ac747b5..645bf51c 100644 --- a/tests/test_documents.py +++ b/tests/test_documents.py @@ -1,8 +1,9 @@ import csv import hashlib from datetime import datetime -from io import StringIO +from io import StringIO, BytesIO from pathlib import Path +from flask import url_for import pytest from werkzeug.exceptions import Unauthorized @@ -463,6 +464,160 @@ def test_get_document_lots(user: UserClient, user2: UserClient): assert export2_csv[1][3] == 'comments,lot3,testcomment-lot3,' +@pytest.mark.mvp +def test_verify_stamp(user: UserClient, client: Client): + """Test verify stamp of one export device information in a csv file.""" + snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot) + csv_str, _ = user.get(res=documents.DocumentDef.t, + item='devices/', + accept='text/csv', + query=[('filter', {'type': ['Computer']})]) + + response, _ = client.post(res=documents.DocumentDef.t, + item='stamps/', + content_type='multipart/form-data', + accept='text/html', + data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), 'example.csv')]}, + status=200) + assert "alert alert-info" in response + assert not "alert alert-danger" in response + + response, _ = client.post(res=documents.DocumentDef.t, + item='stamps/', + content_type='multipart/form-data', + accept='text/html', + data={'docUpload': [(BytesIO(b'abc'), 'example.csv')]}, + status=200) + + assert not "alert alert-info" in response + assert "alert alert-danger" in response + + response, _ = client.get(res=documents.DocumentDef.t, + item='stamps/', + accept='text/html', + status=200) + + assert not "alert alert-info" in response + assert not "alert alert-danger" in response + + +@pytest.mark.mvp +def test_verify_stamp_log_info(user: UserClient, client: Client): + """Test verify stamp of one export lots-info in a csv file.""" + + l, _ = user.post({'name': 'Lot1', 'description': 'comments,lot1,testcomment-lot1,'}, res=Lot) + l, _ = user.post({'name': 'Lot2', 'description': 'comments,lot2,testcomment-lot2,'}, res=Lot) + + csv_str, _ = user.get(res=documents.DocumentDef.t, + item='lots/', + accept='text/csv') + + response, _ = client.post(res=documents.DocumentDef.t, + item='stamps/', + content_type='multipart/form-data', + accept='text/html', + data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), + 'example.csv')]}, + status=200) + assert "alert alert-info" in response + + +@pytest.mark.mvp +def test_verify_stamp_devices_stock(user: UserClient, client: Client): + """Test verify stamp of one export device information in a csv file.""" + + snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot) + + csv_str, _ = user.get(res=documents.DocumentDef.t, + item='stock/', + accept='text/csv', + query=[('filter', {'type': ['Computer']})]) + + response, _ = client.post(res=documents.DocumentDef.t, + item='stamps/', + content_type='multipart/form-data', + accept='text/html', + data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), + 'example.csv')]}, + status=200) + assert "alert alert-info" in response + + +@pytest.mark.mvp +def test_verify_stamp_csv_actions(user: UserClient, client: Client): + """Test verify stamp of one export device information in a csv file with others users.""" + acer = file('acer.happy.battery.snapshot') + snapshot, _ = user.post(acer, res=Snapshot) + device_id = snapshot['device']['id'] + post_request = {"transaction": "ccc", "name": "John", "endUsers": 1, + "devices": [device_id], "description": "aaa", + "finalUserCode": "abcdefjhi", + "startTime": "2020-11-01T02:00:00+00:00", + "endTime": "2020-12-01T02:00:00+00:00" + } + + user.post(res=Allocate, data=post_request) + hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0] + hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0] + hdd_action['lifetime'] += 1000 + acer.pop('elapsed') + acer['licence_version'] = '1.0.0' + snapshot, _ = client.post(acer, res=Live) + + csv_str, _ = user.get(res=documents.DocumentDef.t, + item='actions/', + accept='text/csv', + query=[('filter', {'type': ['Computer']})]) + + response, _ = client.post(res=documents.DocumentDef.t, + item='stamps/', + content_type='multipart/form-data', + accept='text/html', + data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), + 'example.csv')]}, + status=200) + assert "alert alert-info" in response + + +@pytest.mark.mvp +def test_verify_stamp_erasure_certificate(user: UserClient, client: Client): + """Test verify stamp of one export certificate in PDF.""" + s = file('erase-sectors.snapshot') + snapshot, response = user.post(s, res=Snapshot) + # import pdb; pdb.set_trace() + + doc, _ = user.get(res=documents.DocumentDef.t, + item='erasures/', + query=[('filter', {'id': [snapshot['device']['id']]})], + accept=ANY) + + response, _ = client.post(res=documents.DocumentDef.t, + item='stamps/', + content_type='multipart/form-data', + accept='text/html', + data={'docUpload': [(BytesIO(bytes(doc, 'utf-8')), + 'example.csv')]}, + status=200) + assert "alert alert-danger" in response + + doc, _ = user.get(res=documents.DocumentDef.t, + item='erasures/', + query=[ + ('filter', {'id': [snapshot['device']['id']]}), + ('format', 'PDF') + ], + accept='application/pdf') + + response, _ = client.post(res=documents.DocumentDef.t, + item='stamps/', + content_type='multipart/form-data', + accept='text/html', + data={'docUpload': [(BytesIO(doc), + 'example.csv')]}, + status=200) + assert "alert alert-info" in response + + @pytest.mark.mvp def test_get_document_internal_stats(user: UserClient, user2: UserClient): """Tests for get teh internal stats."""