Add document resource and erase certificate

This commit is contained in:
Xavier Bustamante Talavera 2018-11-21 14:26:56 +01:00
parent b59721707d
commit d5a71a7678
21 changed files with 666 additions and 263 deletions

View file

@ -26,6 +26,8 @@ The requirements are:
- PostgreSQL 9.6 or higher with pgcrypto and ltree.
In debian 9 is `# apt install postgresql-contrib`
- passlib. In debian 9 is `# apt install python3-passlib`.
- Weasyprint requires some system packages.
[Their docs explain which ones and how to install them](http://weasyprint.readthedocs.io/en/stable/install.html).
Install Devicehub with *pip*: `pip3 install ereuse-devicehub -U --pre`.

View file

@ -9,6 +9,7 @@ from teal.utils import import_resource
from ereuse_devicehub.resources import agent, event, lot, tag, user
from ereuse_devicehub.resources.device import definitions
from ereuse_devicehub.resources.documents import documents
from ereuse_devicehub.resources.enums import PriceSoftware, RatingSoftware
@ -18,7 +19,9 @@ class DevicehubConfig(Config):
import_resource(user),
import_resource(tag),
import_resource(agent),
import_resource(lot)))
import_resource(lot),
import_resource(documents))
)
PASSWORD_SCHEMES = {'pbkdf2_sha256'} # type: Set[str]
SQLALCHEMY_DATABASE_URI = 'postgresql://dhub:ereuse@localhost/devicehub' # type: str
SCHEMA = 'dhub'

View file

@ -343,7 +343,7 @@ class Computer(Device):
@property
def privacy(self):
"""Returns the privacy of all DataStorage components when
it is None.
it is not None.
"""
return set(
privacy for privacy in
@ -500,7 +500,7 @@ class DataStorage(JoinedComponentTableMixin, Component):
def __format__(self, format_spec):
v = super().__format__(format_spec)
if 's' in format_spec:
v += ' {} GB'.format(self.size // 1000)
v += ' {} GB'.format(self.size // 1000 if self.size else '?')
return v

View file

@ -19,6 +19,7 @@ from ereuse_devicehub.resources.image.models import ImageList
from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.models import Thing
from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.tag.model import Tags
class Device(Thing):
@ -55,7 +56,7 @@ class Device(Thing):
self.events_multiple = ... # type: Set[e.EventWithMultipleDevices]
self.events_one = ... # type: Set[e.EventWithOneDevice]
self.images = ... # type: ImageList
self.tags = ... # type: Set[Tag]
self.tags = ... # type: Tags[Tag]
self.lots = ... # type: Set[Lot]
self.production_date = ... # type: datetime

View file

@ -207,7 +207,9 @@
{{ event._date_str }}
</small>
</div>
{% if event.certificate %}
<a href="{{ event.certificate.to_text() }}">See the certificate</a>
{% endif %}
</li>
{% endfor %}
</ol>

View file

@ -15,7 +15,7 @@ from ereuse_devicehub.query import SearchQueryParser
from ereuse_devicehub.resources import search
from ereuse_devicehub.resources.device.models import Device, Manufacturer
from ereuse_devicehub.resources.device.search import DeviceSearch
from ereuse_devicehub.resources.event.models import Rate
from ereuse_devicehub.resources.event import models as events
from ereuse_devicehub.resources.lot.models import LotDeviceDescendants
from ereuse_devicehub.resources.tag.model import Tag
@ -31,9 +31,9 @@ class OfType(f.Str):
class RateQ(query.Query):
rating = query.Between(Rate.rating, f.Float())
appearance = query.Between(Rate.appearance, f.Float())
functionality = query.Between(Rate.functionality, f.Float())
rating = query.Between(events.Rate.rating, f.Float())
appearance = query.Between(events.Rate.appearance, f.Float())
functionality = query.Between(events.Rate.functionality, f.Float())
class TagQ(query.Query):
@ -46,11 +46,12 @@ class LotQ(query.Query):
class Filters(query.Query):
id = query.Or(query.Equal(Device.id, fields.Integer()))
type = query.Or(OfType(Device.type))
model = query.ILike(Device.model)
manufacturer = query.ILike(Device.manufacturer)
serialNumber = query.ILike(Device.serial_number)
rating = query.Join(Device.id == Rate.device_id, RateQ)
rating = query.Join(Device.id == events.Rate.device_id, RateQ)
tag = query.Join(Device.id == Tag.device_id, TagQ)
# todo This part of the query is really slow
# And forces usage of distinct, as it returns many rows
@ -80,21 +81,13 @@ class DeviceView(View):
parameters:
- name: id
type: integer
in: path
in: path}
description: The identifier of the device.
responses:
200:
description: The device or devices.
"""
# Majority of code is from teal
if id:
response = self.one(id)
else:
args = self.QUERY_PARSER.parse(self.find_args,
request,
locations=('querystring',))
response = self.find(args)
return response
return super().get(id)
def one(self, id: int):
"""Gets one device."""
@ -115,17 +108,8 @@ class DeviceView(View):
@auth.Auth.requires_auth
def find(self, args: dict):
"""Gets many devices."""
search_p = args.get('search', None)
query = Device.query.distinct() # todo we should not force to do this if the query is ok
if search_p:
properties = DeviceSearch.properties
tags = DeviceSearch.tags
query = query.join(DeviceSearch).filter(
search.Search.match(properties, search_p) | search.Search.match(tags, search_p)
).order_by(
search.Search.rank(properties, search_p) + search.Search.rank(tags, search_p)
)
query = query.filter(*args['filter']).order_by(*args['sort'])
# Compute query
query = self.query(args)
devices = query.paginate(page=args['page'], per_page=30) # type: Pagination
ret = {
'items': self.schema.dump(devices.items, many=True, nested=1),
@ -142,6 +126,19 @@ class DeviceView(View):
}
return jsonify(ret)
def query(self, args):
query = Device.query.distinct() # todo we should not force to do this if the query is ok
search_p = args.get('search', None)
if search_p:
properties = DeviceSearch.properties
tags = DeviceSearch.tags
query = query.join(DeviceSearch).filter(
search.Search.match(properties, search_p) | search.Search.match(tags, search_p)
).order_by(
search.Search.rank(properties, search_p) + search.Search.rank(tags, search_p)
)
return query.filter(*args['filter']).order_by(*args['sort'])
class ManufacturerView(View):
class FindArgs(marshmallow.Schema):

View file

@ -0,0 +1,126 @@
import enum
import uuid
from typing import Callable, Iterable, Tuple
import boltons
import flask
import flask_weasyprint
import teal.marshmallow
from boltons import urlutils
from teal.resource import Resource
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.device import models as devs
from ereuse_devicehub.resources.device.views import DeviceView
from ereuse_devicehub.resources.event import models as evs
class Format(enum.Enum):
HTML = 'HTML'
PDF = 'PDF'
class DocumentView(DeviceView):
class FindArgs(DeviceView.FindArgs):
format = teal.marshmallow.EnumField(Format, missing=None)
def get(self, id):
"""Get a collection of resources or a specific one.
---
parameters:
- name: id
in: path
description: The identifier of the resource.
type: string
required: false
responses:
200:
description: Return the collection or the specific one.
"""
args = self.QUERY_PARSER.parse(self.find_args,
flask.request,
locations=('querystring',))
if id:
# todo we assume we can pass both device id and event id
# for certificates... how is it going to end up being?
try:
id = uuid.UUID(id)
except ValueError:
try:
id = int(id)
except ValueError:
raise teal.marshmallow.ValidationError('Document must be an ID or UUID.')
else:
query = devs.Device.query.filter_by(id=id)
else:
query = evs.Event.query.filter_by(id=id)
else:
flask.current_app.auth.requires_auth(lambda: None)() # todo not nice
query = self.query(args)
type = urlutils.URL(flask.request.url).path_parts[-2]
if type == 'erasures':
template = self.erasure(query)
if args.get('format') == Format.PDF:
res = flask_weasyprint.render_pdf(
flask_weasyprint.HTML(string=template), download_filename='{}.pdf'.format(type)
)
else:
res = flask.make_response(template)
return res
@staticmethod
def erasure(query: db.Query):
def erasures():
for model in query:
if isinstance(model, devs.Computer):
for erasure in model.privacy:
yield erasure
elif isinstance(model, devs.DataStorage):
erasure = model.privacy
if erasure:
yield erasure
else:
assert isinstance(model, evs.EraseBasic)
yield model
url_pdf = boltons.urlutils.URL(flask.request.url)
url_pdf.query_params['format'] = 'PDF'
url_web = boltons.urlutils.URL(flask.request.url)
url_web.query_params['format'] = 'HTML'
params = {
'title': 'Erasure Certificate',
'erasures': tuple(erasures()),
'url_pdf': url_pdf.to_text(),
'url_web': url_web.to_text()
}
return flask.render_template('documents/erasure.html', **params)
class DocumentDef(Resource):
__type__ = 'Document'
SCHEMA = None
VIEW = None # We do not want to create default / documents endpoint
AUTH = False
def __init__(self, app,
import_name=__name__,
static_folder='static',
static_url_path=None,
template_folder='templates',
url_prefix=None,
subdomain=None,
url_defaults=None,
root_path=None,
cli_commands: Iterable[Tuple[Callable, str or None]] = tuple()):
super().__init__(app, import_name, static_folder, static_url_path, template_folder,
url_prefix, subdomain, url_defaults, root_path, cli_commands)
d = {'id': None}
get = {'GET'}
view = DocumentView.as_view('main', definition=self, auth=app.auth)
if self.AUTH:
view = app.auth.requires_auth(view)
self.add_url_rule('/erasures/', defaults=d, view_func=view, methods=get)
self.add_url_rule('/erasures/<{}:{}>'.format(self.ID_CONVERTER.value, self.ID_NAME),
view_func=view, methods=get)

View file

@ -0,0 +1,48 @@
/**
Devicehub uses Weasyprint to generate the PDF.
This print.css provides helpful markup to generate the PDF (pages, margins, etc).
The most important things to remember are:
- DOM elements with a class `page-break` create a new page.
- DOM elements with a class `no-page-break` do not break between pages.
- Pages are in A4 by default an 12px.
*/
body {
background-color: transparent !important;
font-size: 12px !important
}
@page {
size: A4;
@bottom-right {
font-family: "Source Sans Pro", Calibri, Candra, Sans serif;
margin-right: 3em;
content: counter(page) " / " counter(pages) !important
}
}
/* Sections produce a new page*/
.page-break:not(section:first-of-type) {
page-break-before: always
}
/* Do not break divs with not-break between pages*/
.no-page-break {
page-break-inside: avoid
}
.print-only, .print-only * {
display: none
}
/* Do not print divs with no-print in them */
@media print {
.no-print, .no-print * {
display: none !important;
}
.print-only, .print-only * {
display: initial;
}
}

View file

@ -0,0 +1,89 @@
{% extends "documents/layout.html" %}
{% block body %}
<div>
<h2>Resumé</h2>
<table class="table table-bordered">
<thead>
<tr>
<th>S/N</th>
<th>Tags</th>
<th>S/N Data Storage</th>
<th>Type of erasure</th>
<th>Result</th>
<th>Date</th>
</tr>
</thead>
<tbody>
{% for erasure in erasures %}
<tr>
<td>
{{ erasure.parent.serial_number.upper() }}
</td>
<td>
{{ erasure.parent.tags }}
</td>
<td>
{{ erasure.device.serial_number.upper() }}
</td>
<td>
{{ erasure.type }}
</td>
<td>
{{ erasure.severity }}
</td>
<td>
{{ erasure.date_str }}
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
<div class="page-break row">
<h2>Details</h2>
{% for erasure in erasures %}
<div class="col-md-6 no-page-break">
<h4>{{ erasure.device.__format__('t') }}</h4>
<dl>
<dt>Data storage:</dt>
<dd>{{ erasure.device.__format__('ts') }}</dd>
<dt>Computer:</dt>
<dd>{{ erasure.parent.__format__('ts') }}</dd>
<dt>Tags:</dt>
<dd>{{ erasure.parent.tags }}</dd>
<dt>Erasure:</dt>
<dd>{{ erasure.__format__('ts') }}</dd>
<dt>Erasure steps:</dt>
<dd>
<ol>
{% for step in erasure.steps %}
<li>{{ step.__format__('') }}</li>
{% endfor %}
</ol>
</dd>
</dl>
</div>
{% endfor %}
</div>
<div class="no-page-break">
<h2>Glossary</h2>
<dl>
<dt>Erase Basic</dt>
<dd>
A software-based fast non-100%-secured way of erasing data storage,
using <a href="https://en.wikipedia.org/wiki/Shred_(Unix)">shred</a>.
</dd>
<dt>Erase Sectors</dt>
<dd>
A secured-way of erasing data storages, checking sector-by-sector
the erasure, using <a href="https://en.wikipedia.org/wiki/Badblocks">badblocks</a>.
</dd>
</dl>
</div>
<div class="no-print">
<a href="{{ url_pdf }}">Click here to download the PDF.</a>
</div>
<div class="print-only">
<a href="{{ url_web }}">Verify on-line the integrity of this document</a>
</div>
{% endblock %}

View file

@ -0,0 +1,26 @@
{% import 'devices/macros.html' as macros %}
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
<link href="https://stackpath.bootstrapcdn.com/bootswatch/3.3.7/flatly/bootstrap.min.css"
rel="stylesheet"
integrity="sha384-+ENW/yibaokMnme+vBLnHMphUYxHs34h9lpdbSLuAwGkOKFRl4C34WkjazBtb7eT"
crossorigin="anonymous">
<link rel="stylesheet"
type="text/css"
href="{{ url_for('Document.static', filename='print.css') }}">
<title>Devicehub | {{ title }}</title>
</head>
<body>
<div class="container">
<div class="row">
<header class="page-header">
<h1> {{ title }}</h1>
</header>
</div>
{% block body %}{% endblock %}
</div>
</body>
</html>

View file

@ -310,6 +310,9 @@ class Severity(IntEnum):
m = ''
return m
def __format__(self, format_spec):
return str(self)
class PhysicalErasureMethod(Enum):
"""Methods of physically erasing the data-storage, usually

View file

@ -2,7 +2,7 @@ from collections import Iterable
from datetime import datetime, timedelta
from decimal import Decimal, ROUND_HALF_EVEN, ROUND_UP
from distutils.version import StrictVersion
from typing import Set, Union
from typing import Optional, Set, Union
from uuid import uuid4
import inflection
@ -157,11 +157,21 @@ class Event(Thing):
would point to the computer that contained this data storage, if any.
"""
@property
def elapsed(self):
"""Returns the elapsed time with seconds precision."""
t = self.end_time - self.start_time
return timedelta(seconds=t.seconds)
@property
def url(self) -> urlutils.URL:
"""The URL where to GET this event."""
return urlutils.URL(url_for_resource(Event, item_id=self.id))
@property
def certificate(self) -> Optional[urlutils.URL]:
return None
# noinspection PyMethodParameters
@declared_attr
def __mapper_args__(cls):
@ -193,7 +203,7 @@ class Event(Thing):
return start_time
@property
def _date_str(self):
def date_str(self):
return '{:%c}'.format(self.end_time or self.created)
def __str__(self) -> str:
@ -311,20 +321,44 @@ class EraseBasic(JoinedWithOneDeviceMixin, EventWithOneDevice):
Devicehub automatically shows the standards that each erasure
follows.
"""
method = 'Shred'
"""The method or software used to destroy the data."""
@property
def standards(self):
"""A set of standards that this erasure follows."""
return ErasureStandards.from_data_storage(self)
@property
def certificate(self):
"""The URL of this erasure certificate."""
# todo will this url_for_resoure work for other resources?
return urlutils.URL(url_for_resource('Document', item_id=self.id))
def __str__(self) -> str:
return '{} on {}.'.format(self.severity, self.end_time)
return '{} on {}.'.format(self.severity, self.date_str)
def __format__(self, format_spec: str) -> str:
v = ''
if 't' in format_spec:
v += '{} {}'.format(self.type, self.severity)
if 't' in format_spec and 's' in format_spec:
v += '. '
if 's' in format_spec:
if self.standards:
std = 'with standards {}'.format(self.standards)
else:
std = 'no standard'
v += 'Method used: {}, {}. '.format(self.method, std)
v += '{} elapsed, on {}'.format(self.elapsed, self.date_str)
return v
class EraseSectors(EraseBasic):
"""A secured-way of erasing data storages, checking sector-by-sector
the erasure, using `badblocks <https://en.wikipedia.org/wiki/Badblocks>`_.
"""
method = 'Badblocks'
class ErasePhysical(EraseBasic):
@ -348,6 +382,12 @@ class Step(db.Model):
order_by=num,
collection_class=ordering_list('num')))
@property
def elapsed(self):
"""Returns the elapsed time with seconds precision."""
t = self.end_time - self.start_time
return timedelta(seconds=t.seconds)
# noinspection PyMethodParameters
@declared_attr
def __mapper_args__(cls):
@ -363,6 +403,9 @@ class Step(db.Model):
args[POLYMORPHIC_ON] = cls.type
return args
def __format__(self, format_spec: str) -> str:
return '{} {} {}'.format(self.severity, self.type, self.elapsed)
class StepZero(Step):
pass

View file

@ -2,7 +2,7 @@ import ipaddress
from datetime import datetime, timedelta
from decimal import Decimal
from distutils.version import StrictVersion
from typing import Dict, List, Set, Union
from typing import Dict, List, Optional, Set, Union
from uuid import UUID
from boltons import urlutils
@ -358,6 +358,10 @@ class EraseBasic(EventWithOneDevice):
def standards(self) -> Set[ErasureStandards]:
pass
@property
def certificate(self) -> urlutils.URL:
pass
class EraseSectors(EraseBasic):
def __init__(self, **kwargs) -> None:

View file

@ -1,4 +1,5 @@
from contextlib import suppress
from typing import Set
from sqlalchemy import BigInteger, Column, ForeignKey, Unicode, UniqueConstraint
from sqlalchemy.dialects.postgresql import UUID
@ -11,6 +12,14 @@ from ereuse_devicehub.resources.device.models import Device
from ereuse_devicehub.resources.models import Thing
class Tags(Set['Tag']):
def __str__(self) -> str:
return ', '.join(str(tag) for tag in self).strip()
def __format__(self, format_spec):
return ', '.join(format(tag, format_spec) for tag in self).strip()
class Tag(Thing):
id = Column(Unicode(), check_lower('id'), primary_key=True)
id.comment = """The ID of the tag."""
@ -35,7 +44,7 @@ class Tag(Thing):
ForeignKey(Device.id, ondelete=DB_CASCADE_SET_NULL),
index=True)
device = relationship(Device,
backref=backref('tags', lazy=True, collection_class=set),
backref=backref('tags', lazy=True, collection_class=Tags),
primaryjoin=Device.id == device_id)
"""The device linked to this tag."""
secondary = Column(Unicode(), check_lower('secondary'), index=True)
@ -82,3 +91,9 @@ class Tag(Thing):
def __repr__(self) -> str:
return '<Tag {0.id} org:{0.org_id} device:{0.device_id}>'.format(self)
def __str__(self) -> str:
return '{0.id} org: {0.org.name} device: {0.device}'.format(self)
def __format__(self, format_spec: str) -> str:
return '{0.org.name} {0.id}'

View file

@ -29,3 +29,5 @@ teal==0.2.0a30
webargs==4.0.0
Werkzeug==0.14.1
sqlalchemy-citext==1.3.post0
flask-weasyprint==0.5
weasyprint==43

View file

@ -42,6 +42,7 @@ setup(
'requests-toolbelt',
'sqlalchemy-citext',
'sqlalchemy-utils[password, color, phone]',
'Flask-WeasyPrint'
],
extras_require={
'docs': [

View file

@ -28,6 +28,8 @@ def test_api_docs(client: Client):
'/manufacturers/',
'/lots/{id}/children',
'/lots/{id}/devices',
'/documents/erasures/',
'/documents/static/{filename}',
'/tags/{tag_id}/device/{device_id}',
'/devices/static/{filename}'
}

View file

@ -184,11 +184,6 @@ def test_device_query(user: UserClient):
assert not pc['tags']
@pytest.mark.xfail(reason='Functionality not yet developed.')
def test_device_lots_query(user: UserClient):
pass
def test_device_search_all_devices_token_if_empty(app: Devicehub, user: UserClient):
"""Ensures DeviceSearch can regenerate itself when the table is empty."""
user.post(file('basic.snapshot'), res=Snapshot)

65
tests/test_documents.py Normal file
View file

@ -0,0 +1,65 @@
import teal.marshmallow
from ereuse_utils.test import ANY
from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.resources.documents import documents as docs
from ereuse_devicehub.resources.event import models as e
from tests.conftest import file
def test_erasure_certificate_public_one(user: UserClient, client: Client):
"""Public user can get certificate from one device as HTML or PDF."""
s = file('erase-sectors.snapshot')
snapshot, _ = user.post(s, res=e.Snapshot)
doc, response = client.get(res=docs.DocumentDef.t,
item='erasures/{}'.format(snapshot['device']['id']),
accept=ANY)
assert 'html' in response.content_type
assert '<html' in doc
assert '2018' in doc
doc, response = client.get(res=docs.DocumentDef.t,
item='erasures/{}'.format(snapshot['device']['id']),
query=[('format', 'PDF')],
accept='application/pdf')
assert 'application/pdf' == response.content_type
erasure = next(e for e in snapshot['events'] if e['type'] == 'EraseSectors')
doc, response = client.get(res=docs.DocumentDef.t,
item='erasures/{}'.format(erasure['id']),
accept=ANY)
assert 'html' in response.content_type
assert '<html' in doc
assert '2018' in doc
def test_erasure_certificate_private_query(user: UserClient):
"""Logged-in user can get certificates using queries as HTML and
PDF.
"""
s = file('erase-sectors.snapshot')
snapshot, response = user.post(s, res=e.Snapshot)
doc, response = user.get(res=docs.DocumentDef.t,
item='erasures/',
query=[('filter', {'id': [snapshot['device']['id']]})],
accept=ANY)
assert 'html' in response.content_type
assert '<html' in doc
assert '2018' in doc
doc, response = user.get(res=docs.DocumentDef.t,
item='erasures/',
query=[
('filter', {'id': [snapshot['device']['id']]}),
('format', 'PDF')
],
accept='application/pdf')
assert 'application/pdf' == response.content_type
def test_erasure_certificate_wrong_id(client: Client):
client.get(res=docs.DocumentDef.t, item='erasures/this-is-not-an-id',
status=teal.marshmallow.ValidationError)

View file

@ -351,27 +351,6 @@ def test_erase_physical():
db.session.commit()
@pytest.mark.xfail(reson='validate use-case')
def test_view_public_erasure_certificate():
"""User can see html erasure certificate even if not logged-in,
from the public link.
"""
@pytest.mark.xfail(reson='Validate use-case')
def test_not_download_erasure_certificate_if_public():
"""User cannot download an erasure certificate as PDF if
not logged-in.
"""
@pytest.mark.xfail(reson='talk to Jordi about variables in certificate erasure.')
def test_download_erasure_certificate():
"""User can download erasure certificates. We test erasure
certificates with: ... todo
"""
@pytest.mark.xfail(reson='Adapt rate algorithm to re-compute by passing a manual rate.')
def test_manual_rate_after_workbench_rate(user: UserClient):
"""Perform a WorkbenchRate and then update the device with a ManualRate.