First commit: create mvp project
- create an orchestrator to orchestrate signature of pdf; - create a worker to get request from amqp and resend it to amqp
This commit is contained in:
@@ -0,0 +1,74 @@
|
||||
import io
|
||||
from typing import Optional
|
||||
|
||||
from pyhanko import stamp
|
||||
from pyhanko.pdf_utils.incremental_writer import IncrementalPdfFileWriter
|
||||
from pyhanko.sign import signers, timestamps, fields
|
||||
from pyhanko_certvalidator import ValidationContext
|
||||
from typing_extensions import Buffer
|
||||
|
||||
|
||||
class SignOrchestrator:
|
||||
"""Orchestrate the signature on document"""
|
||||
|
||||
def __init__(self, pkcs12_path: str, timestamp_url: str, pkcs12_password: Optional[bytes] = None):
|
||||
# Load signer key material from PKCS#12 file
|
||||
# This assumes that any relevant intermediate certs are also included
|
||||
# in the PKCS#12 file.
|
||||
self.signer = signers.SimpleSigner.load_pkcs12(
|
||||
pfx_file=pkcs12_path, passphrase=pkcs12_password
|
||||
)
|
||||
|
||||
# Set up a timestamping client to fetch timestamps tokens
|
||||
self.timestamper = timestamps.HTTPTimeStamper(
|
||||
url=timestamp_url,
|
||||
)
|
||||
|
||||
self.stamp_style = stamp.TextStampStyle(
|
||||
stamp_text="Signé par:\n%(signer_text)s\nLe %(ts)s",
|
||||
border_width=1,
|
||||
)
|
||||
|
||||
|
||||
def _make_signature_metadata(self, reason: str, field_name: str):
|
||||
# Settings for PAdES-LTA
|
||||
return signers.PdfSignatureMetadata(
|
||||
field_name=field_name, md_algorithm='sha256',
|
||||
# Mark the signature as a PAdES signature
|
||||
subfilter=fields.SigSeedSubFilter.PADES,
|
||||
# We'll also need a validation context
|
||||
# to fetch & embed revocation info.
|
||||
# validation_context=ValidationContext(allow_fetching=True),
|
||||
# Embed relevant OCSP responses / CRLs (PAdES-LT)
|
||||
# embed_validation_info=True,
|
||||
# Tell pyHanko to put in an extra DocumentTimeStamp
|
||||
# to kick off the PAdES-LTA timestamp chain.
|
||||
use_pades_lta=True,
|
||||
reason=reason,
|
||||
)
|
||||
|
||||
def sign(self, reason: str, signature_index: int, input_content: Buffer, on_page: int, box_place: (int, int, int, int), signer_text: str) -> io.BytesIO:
|
||||
field_name = 'Signature' + str(signature_index)
|
||||
signature_meta = self._make_signature_metadata(reason, field_name)
|
||||
|
||||
pdf_signer = signers.PdfSigner(
|
||||
signature_meta, signer=self.signer, timestamper=self.timestamper,
|
||||
stamp_style=self.stamp_style
|
||||
)
|
||||
|
||||
inf = io.BytesIO(input_content)
|
||||
w = IncrementalPdfFileWriter(inf)
|
||||
fields.append_signature_field(w, sig_field_spec=fields.SigFieldSpec(
|
||||
field_name, on_page=on_page, box=box_place
|
||||
))
|
||||
outf = io.BytesIO()
|
||||
pdf_signer.sign_pdf(
|
||||
w, output=outf, appearance_text_params={'signer_text': signer_text}
|
||||
)
|
||||
|
||||
return outf
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user