1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
import base64
import binascii
import datetime
import io
import subprocess
try:
from defusedxml import ElementTree as ET
except ModuleNotFoundError:
import xml.etree.ElementTree as ET
from typing import BinaryIO, Union
from asn1crypto.cms import ContentInfo
from . import fattura as a38
class SignatureVerificationError(Exception):
pass
class InvalidSignatureError(SignatureVerificationError):
pass
class SignerCertificateError(SignatureVerificationError):
pass
class P7M:
"""
Parse a Fattura Elettronica encoded as a .p7m file
"""
def __init__(self, data: Union[str, bytes, BinaryIO]):
"""
If data is a string, it is taken as a file name.
If data is bytes, it is taken as p7m data.
Otherwise, data is taken as a file-like object that reads bytes data.
"""
if isinstance(data, str):
with open(data, "rb") as fd:
self.data = fd.read()
elif isinstance(data, bytes):
self.data = data
else:
self.data = data.read()
# Data might potentially be base64 encoded
try:
self.data = base64.b64decode(self.data, validate=True)
except binascii.Error:
pass
self.content_info = ContentInfo.load(self.data)
def is_expired(self) -> bool:
"""
Check if the signature has expired
"""
now = datetime.datetime.utcnow()
signed_data = self.get_signed_data()
for c in signed_data["certificates"]:
if c.name != "certificate":
# The signatures I've seen so far use 'certificate' only
continue
expiration_date = c.chosen["tbs_certificate"]["validity"]["not_after"].chosen.native.replace(tzinfo=None)
if expiration_date <= now:
return True
return False
def get_signed_data(self):
"""
Return the SignedData part of the P7M file
"""
if self.content_info["content_type"].native != "signed_data":
raise RuntimeError("p7m data is not an instance of signed_data")
signed_data = self.content_info["content"]
if signed_data["version"].native != "v1":
raise RuntimeError(f"ContentInfo/SignedData.version is {signed_data['version'].native} instead of v1")
return signed_data
def get_payload(self):
"""
Return the raw signed data
"""
signed_data = self.get_signed_data()
encap_content_info = signed_data["encap_content_info"]
return encap_content_info["content"].native
def get_fattura(self):
"""
Return the parsed XML data
"""
data = io.BytesIO(self.get_payload())
tree = ET.parse(data)
return a38.auto_from_etree(tree.getroot())
def verify_signature(self, certdir):
"""
Verify the signature on the file
"""
res = subprocess.run([
"openssl", "cms", "-verify", "-inform", "DER", "-CApath", certdir, "-noout"],
input=self.data,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE)
# From openssl cms manpage:
# 0 The operation was completely successfully.
# 1 An error occurred parsing the command options.
# 2 One of the input files could not be read.
# 3 An error occurred creating the CMS file or when reading the MIME message.
# 4 An error occurred decrypting or verifying the message.
# 5 The message was verified correctly but an error occurred writing out the signers certificates.
if res.returncode == 0:
pass
elif res.returncode == 4:
raise InvalidSignatureError(res.stderr)
elif res.returncode == 5:
raise SignerCertificateError(res.stderr)
else:
raise RuntimeError(res.stderr)
|
