Newer
Older
# Copyright 2019 PrivateStorage.io, LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
A module for logic controlling the manner in which ZKAPs are spent.
"""
from __future__ import absolute_import, division, print_function, unicode_literals
from future.utils import PY2
if PY2:
from future.builtins import ( # noqa: F401
filter,
map,
zip,
ascii,
chr,
hex,
input,
next,
oct,
open,
pow,
round,
super,
bytes,
dict,
list,
object,
range,
str,
max,
min,
)
import attr
from .eliot import GET_PASSES, INVALID_PASSES, RESET_PASSES, SPENT_PASSES
class IPassGroup(Interface):
"""
A group of passed meant to be spent together.
"""
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
passes = Attribute(":ivar list[Pass] passes: The passes themselves.")
def split(select_indices):
"""
Create two new ``IPassGroup`` providers. The first contains all passes in
this group at the given indices. The second contains all the others.
:param list[int] select_indices: The indices of the passes to include
in the first resulting group.
:return (IPassGroup, IPassGroup): The two new groups.
"""
def expand(by_amount):
"""
Create a new ``IPassGroup`` provider which contains all of this group's
passes and some more.
:param int by_amount: The number of additional passes the resulting
group should contain.
:return IPassGroup: The new group.
"""
def mark_spent():
"""
The passes have been spent successfully. Ensure none of them appear in
any ``IPassGroup`` provider created in the future.
:return: ``None``
"""
def mark_invalid(reason):
"""
The passes could not be spent. Ensure none of them appear in any
``IPassGroup`` provider created in the future.
:param unicode reason: A short description of the reason the passes
could not be spent.
:return: ``None``
"""
def reset():
"""
The passes have not been spent. Return them to for use in a future
``IPassGroup`` provider.
:return: ``None``
"""
class IPassFactory(Interface):
"""
An object which can create passes.
"""
def get(message, num_passes):
"""
:param unicode message: A request-binding message for the resulting passes.
:param int num_passes: The number of passes to request.
:return IPassGroup: A group of passes bound to the given message and
of the requested size.
"""
@implementer(IPassGroup)
@attr.s
class PassGroup(object):
"""
Track the state of a group of passes intended as payment for an operation.
:ivar _message: The request binding message for this group of
passes.
:ivar IPassFactory _factory: The factory which created this pass group.
:ivar list[Pass] passes: The passes of which this group consists.
"""
_message = attr.ib(validator=attr.validators.instance_of(bytes)) # type: bytes
_factory = attr.ib(validator=attr.validators.provides(IPassFactory)) # type: IPassFactory
_tokens = attr.ib(validator=attr.validators.instance_of(list)) # type: List[(UnblinidedToken, Pass)]
@property
def passes(self):
# type: () -> List[Pass]
return list(pass_ for (unblinded_token, pass_) in self._tokens)
@property
def unblinded_tokens(self):
# type: () -> List[UnblindedToken]
return list(unblinded_token for (unblinded_token, pass_) in self._tokens)
def split(self, select_indices):
# type: (List[int]) -> (PassGroup, PassGroup)
selected = []
unselected = []
for idx, t in enumerate(self._tokens):
if idx in select_indices:
selected.append(t)
unselected.append(t)
attr.evolve(self, tokens=selected),
attr.evolve(self, tokens=unselected),
)
def expand(self, by_amount):
# type: (int) -> PassGroup
return attr.evolve(
self,
tokens=self._tokens + self._factory.get(self._message, by_amount)._tokens,
)
def mark_spent(self):
# type: () -> None
self._factory._mark_spent(self.unblinded_tokens)
def mark_invalid(self, reason):
# type: () -> None
self._factory._mark_invalid(reason, self.unblinded_tokens)
def reset(self):
self._factory._reset(self.unblinded_tokens)
@attr.s
class SpendingController(object):
"""
A ``SpendingController`` gives out ZKAPs and arranges for re-spend
attempts when necessary.
"""
get_unblinded_tokens = attr.ib() # type: (int) -> [UnblindedToken]
discard_unblinded_tokens = attr.ib() # type: ([UnblindedTokens]) -> None
invalidate_unblinded_tokens = attr.ib() # type: ([UnblindedTokens]) -> None
reset_unblinded_tokens = attr.ib() # type: ([UnblindedTokens]) -> None
tokens_to_passes = attr.ib() # type: (bytes, [UnblindedTokens]) -> [Pass]
@classmethod
def for_store(cls, tokens_to_passes, store):
return cls(
get_unblinded_tokens=store.get_unblinded_tokens,
discard_unblinded_tokens=store.discard_unblinded_tokens,
invalidate_unblinded_tokens=store.invalidate_unblinded_tokens,
reset_unblinded_tokens=store.reset_unblinded_tokens,
tokens_to_passes=tokens_to_passes,
)
def get(self, message, num_passes):
unblinded_tokens = self.get_unblinded_tokens(num_passes)
passes = self.tokens_to_passes(message, unblinded_tokens)
GET_PASSES.log(
message=message.decode("utf-8"),
count=num_passes,
)
return PassGroup(message, self, list(zip(unblinded_tokens, passes)))
def _mark_spent(self, unblinded_tokens):
SPENT_PASSES.log(
count=len(unblinded_tokens),
)
self.discard_unblinded_tokens(unblinded_tokens)
def _mark_invalid(self, reason, unblinded_tokens):
INVALID_PASSES.log(
reason=reason,
count=len(unblinded_tokens),
)
self.invalidate_unblinded_tokens(reason, unblinded_tokens)
def _reset(self, unblinded_tokens):
RESET_PASSES.log(
count=len(unblinded_tokens),
)
self.reset_unblinded_tokens(unblinded_tokens)