Merge pull request #122 from eReuse/feature/#121-check-stamps

Feature/#121 check stamps

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

ereuse_devicehub/__init__.py

@@ -1 +1 @@
-__version__ = "1.0.4-beta"
+__version__ = "1.0.5-beta"

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 url.to_text() 
-        return flask.render_template('documents/stamp.html', rq_url=url_path)
+
+    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)

ereuse_devicehub/resources/documents/templates/documents/stamp.html

@@ -38,6 +38,17 @@
 </nav>
 <div class="container-fluid">
     <div class="page-header col-md-6 col-md-offset-3">
+      <div class="row">
+      {% if result.0 == 'Ok' %}
+        <div class="alert alert-info" style="background-color: #3fb618;" role="alert">
+	  {{ result.1 }}
+        </div>
+      {% elif result.0 == 'Bad' %}
+        <div class="alert alert-danger" style="background-color: #ff0039" role="alert">
+	  {{ result.1 }}
+        </div>
+      {% endif %}
+      </div>
       <div class="row">
 	      <a href="http://dlt.ereuse.org/stamps/create?url={{ rq_url }}" target="_blank">Add one new check in your csv</a>
       </div>
@@ -45,6 +56,19 @@
 	      <a href="http://dlt.ereuse.org/stamps/check?url={{ rq_url }}" target="_blank">Verify a CSV file in here.</a>
       </div>
     </div>
+
+    <div class="row">
+      <div class="page-header col-md-6 col-md-offset-3">
+	If you want us to verify a document issued by us, upload it using the following form
+      </div>
+      <div class="col-md-6 col-md-offset-3">
+      <form enctype="multipart/form-data" action="" method="post">
+        <label for="name">Document: </label>
+        <input type="file" name="docUpload" accept="*" /><br />
+        <input type="submit" id="send-signup" name="signup" value="Verify" />
+      </form>
+      </div>
+    </div>
 </div>
 </body>
 </html>

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()

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."""