Merge branch 'main' into ricardo-email-report

src/authentication.py

@@ -1,4 +1,5 @@
-from src.database import Users, IntegrityError, DoesNotExist
+from datetime import datetime, timedelta
+from src.database import Users, Sessions, IntegrityError, DoesNotExist
 from src.helpers import string_in_range
 from src.error import InputError
 from src.type_structure import *
@@ -9,7 +10,7 @@
 Auth login and register functions.
 '''
 
-def auth_login_v1(email, password):
+def auth_login_v1(email, password) -> AuthReturnV1:
     '''
     This function uses user-inputted login data to log a user into the system.
 
@@ -29,18 +30,15 @@ def auth_login_v1(email, password):
     try:
         user = Users.get(email=email)
     except DoesNotExist:
-        raise InputError(detail="Invalid input: No user with email " + email + ".")
+        raise InputError(detail="Invalid input: Incorrect email or password.")
 
     if user.password_hash != hashlib.sha256(password.encode("utf-8")).hexdigest():
-        raise InputError(detail="Invalid input: Incorrect password.")
+        raise InputError(detail="Invalid input: Incorrect email or password.")
 
+    return AuthReturnV1(auth_user_id=user.id)
 
-    # data_store.start_token_session(data_store.encode_user_jwt(user_id))
 
-    return {'auth_user_id' : user.id}
-
-
-def auth_register_v1(email, password):
+def auth_register_v1(email, password) -> AuthReturnV1:
     '''
     This function registers a user into the system by getting their details.
 
@@ -71,7 +69,6 @@ def auth_register_v1(email, password):
         raise InputError(detail="Invalid input: Password is too short.")
 
     # Generate password hash
-    # salt = data_store.gen_salt()
     password_hash = hashlib.sha256(password.encode("utf-8")).hexdigest()
 
     try:
@@ -81,6 +78,20 @@ def auth_register_v1(email, password):
         raise InputError(detail="Invalid input: Email " + email + " is already taken.")
 
     # Return id once register is successful
-    return {
-        'auth_user_id': user.id,
-    }
+    return AuthReturnV1(auth_user_id=user.id)
+
+
+def auth_login_v2(email, password) -> AuthReturnV2:
+    id = auth_login_v1(email, password).auth_user_id
+    now = datetime.now()
+    token = hashlib.sha256(id.to_bytes(8, 'big') + now.strftime("%s").encode("utf-8")).hexdigest()
+    Sessions.create(user=id, token=token, date_created=now, date_expires=now + timedelta(days=1))
+    return AuthReturnV2(token=token)
+
+
+def auth_register_v2(email, password) -> AuthReturnV2:
+    id = auth_register_v1(email, password).auth_user_id
+    now = datetime.now()
+    token = hashlib.sha256(id.to_bytes(8, 'big') + now.strftime("%s").encode("utf-8")).hexdigest()
+    Sessions.create(user=id, token=token, date_created=now, date_expires=now + timedelta(days=1))
+    return AuthReturnV2(token=token)

src/database.py

@@ -55,6 +55,7 @@ class Reports(BaseModel):
     schema = ForeignKeyField(Evaluations, backref='schema', null=True, default=None)
     syntax = ForeignKeyField(Evaluations, backref='syntax', null=True, default=None)
     peppol = ForeignKeyField(Evaluations, backref='peppol', null=True, default=None)
+    owner = ForeignKeyField(Users, backref='users', null=True)
 
     def to_json(self):
         return {

src/error.py

@@ -5,9 +5,14 @@ def __init__(self, detail: str):
         self.status_code = 400
         self.detail = detail
 
-class TokenError(Exception):
+class UnauthorisedError(Exception):
     def __init__(self, detail: str):
-        self.status_code = 402
+        self.status_code = 401
+        self.detail = detail
+
+class ForbiddenError(Exception):
+    def __init__(self, detail: str):
+        self.status_code = 403
         self.detail = detail
 
 class NotFoundError(Exception):

src/export.py

@@ -9,7 +9,7 @@
 from src.error import *
 
 
-def export_json_report_v1(report_id: int):
+def export_json_report_v1(report_id: int, owner=None):
     if report_id < 0:
         raise InputError(detail="Report id cannot be less than 0")
 
@@ -18,9 +18,12 @@ def export_json_report_v1(report_id: int):
     except DoesNotExist:
         raise NotFoundError(detail=f"Report with id {report_id} not found")
 
+    if report.owner != None and report.owner != owner:
+        raise ForbiddenError("You do not have permission to view report")
+
     return Report(**report.to_json())
 
-def export_pdf_report_v1(report_id: int) -> bytes:
+def export_pdf_report_v1(report_id: int, owner=None) -> bytes:
     if report_id < 0:
         raise InputError(detail="Report id cannot be less than 0")
 
@@ -29,6 +32,9 @@ def export_pdf_report_v1(report_id: int) -> bytes:
     except DoesNotExist:
         raise NotFoundError(detail=f"Report with id {report_id} not found")
 
+    if report.owner != None and report.owner != owner:
+        raise ForbiddenError("You do not have permission to view report")
+
     html = export_html_report_v1(report_id)
     pdf_bytes = HTML(string=html).write_pdf()
 
@@ -73,7 +79,7 @@ def add_violations(soup, violations, parent):
             location_string = "Line " + str(violation["line"]) + ", Column " + str(violation["column"])
         v.find("code", {"name": "location"}).string = location_string
 
-def export_html_report_v1(report_id: int):
+def export_html_report_v1(report_id: int, owner=None):
     if report_id < 0:
         raise InputError(detail="Report id cannot be less than 0")
 
@@ -82,6 +88,9 @@ def export_html_report_v1(report_id: int):
     except DoesNotExist:
         raise NotFoundError(detail=f"Report with id {report_id} not found")
 
+    if report.owner != None and report.owner != owner:
+        raise ForbiddenError("You do not have permission to view report")
+
     report = report.to_json()
 
     with open("src/report_template.html", "r") as file:
@@ -152,7 +161,7 @@ def write_violations(writer, violations):
         ]
         writer.writerow(data)
 
-def export_csv_report_v1(report_id: int):
+def export_csv_report_v1(report_id: int, owner=None):
     if report_id < 0:
         raise InputError(detail="Report id cannot be less than 0")
 
@@ -161,6 +170,9 @@ def export_csv_report_v1(report_id: int):
     except DoesNotExist:
         raise NotFoundError(detail=f"Report with id {report_id} not found")
 
+    if report.owner != None and report.owner != owner:
+        raise ForbiddenError("You do not have permission to view report")
+
     report = report.to_json()
 
     csv_buffer = StringIO()
@@ -185,17 +197,15 @@ def export_csv_report_v1(report_id: int):
 
     return csv_contents
 
-def report_bulk_export_json_v1(report_ids) -> List:
-    return {
-        "reports": [export_json_report_v1(report_id) for report_id in report_ids]
-    }
+def report_bulk_export_json_v1(report_ids, owner=None) -> ReportList:
+    return ReportList(reports=[export_json_report_v1(report_id, owner) for report_id in report_ids])
 
-def report_bulk_export_pdf_v1(report_ids) -> BytesIO:
+def report_bulk_export_pdf_v1(report_ids, owner=None) -> BytesIO:
     reports = BytesIO()
 
     with ZipFile(reports, 'w', ZIP_DEFLATED) as f:
         for report_id in report_ids:
-            f.writestr(f"invoice_validation_report_{report_id}.pdf", export_pdf_report_v1(report_id))
+            f.writestr(f"invoice_validation_report_{report_id}.pdf", export_pdf_report_v1(report_id, owner))
 
     reports.seek(0)
 

src/generation.py

@@ -18,14 +18,14 @@ def generate_schema_evaluation(invoice_text: str) -> Evaluations:
 
 def generate_parser_evaluation(violations) -> Evaluations:
     evaluation = Evaluations.create(
-        is_valid=True if len(violations) == 0 else False,
+        is_valid=len(violations) == 0,
         num_warnings=0,
         num_errors=len(violations),
         num_rules_failed=len(violations)
     )
 
     for violation in violations:
-        violation.evaluation = evaluation.id 
+        violation.evaluation = evaluation.id #type: ignore
         violation.save()
 
     return evaluation
@@ -52,7 +52,7 @@ def generate_xslt_evaluation(executable, invoice_text) -> Evaluations:
 
     return evaluation
 
-def generate_report(invoice_name: str, invoice_text: str) -> int:
+def generate_report(invoice_name: str, invoice_text: str, owner) -> int:
     wellformedness_evaluation = None
     schema_evaluation = None
     syntax_evaluation = None
@@ -89,12 +89,13 @@ def generate_report(invoice_name: str, invoice_text: str) -> int:
         wellformedness=wellformedness_evaluation.id if wellformedness_evaluation else None,
         schema=schema_evaluation.id if schema_evaluation else None,
         syntax=syntax_evaluation.id if syntax_evaluation else None,
-        peppol=peppol_evaluation.id if peppol_evaluation else None
+        peppol=peppol_evaluation.id if peppol_evaluation else None,
+        owner=owner
     )
 
     return report.id
 
-def generate_diagnostic_list(invoice_text: str) -> int:
+def generate_diagnostic_list(invoice_text: str) -> List[LintDiagnostic]:
     report = []
 
     wellformedness_violations = get_wellformedness_violations(invoice_text)
@@ -106,6 +107,7 @@ def generate_diagnostic_list(invoice_text: str) -> int:
             column=violation.column,
             xpath=violation.xpath,
             message=violation.message,
+            suggestion=None,
             severity="error" if violation.is_fatal else "warning"
         ))
 
@@ -121,6 +123,7 @@ def generate_diagnostic_list(invoice_text: str) -> int:
             column=violation.column,
             xpath=violation.xpath,
             message=violation.message,
+            suggestion=None,
             severity="error" if violation.is_fatal else "warning"
         ))
 

src/invoice.py

@@ -5,16 +5,14 @@
 from src.generation import generate_report
 
 
-def invoice_upload_text_v1(invoice_name: str, invoice_text: str):
+def invoice_upload_text_v1(invoice_name: str, invoice_text: str, owner = None) -> ReportID:
     if len(invoice_name) > 100:
         raise InputError(detail="Name cannot be longer than 100 characters")
 
-    return {
-        "report_id": generate_report(invoice_name, invoice_text)
-    }
+    return ReportID(report_id=generate_report(invoice_name, invoice_text, owner))
 
 
-def invoice_upload_url_v1(invoice_name: str, invoice_url: str):
+def invoice_upload_url_v1(invoice_name: str, invoice_url: str, owner = None):
     if len(invoice_name) > 100:
         raise InputError(detail="Name cannot be longer than 100 characters")
 
@@ -27,21 +25,15 @@ def invoice_upload_url_v1(invoice_name: str, invoice_url: str):
         raise InputError(detail="URL does not point to plain text or XML data")
 
     invoice_text = response.text
-
-    report_id = generate_report(invoice_name, invoice_text)
 
-    return {
-        "report_id": report_id
-    }
+    return ReportID(report_id=generate_report(invoice_name, invoice_text, owner))
 
 
-def invoice_upload_file_v1(invoice_name: str, invoice_text: str):
+def invoice_upload_file_v1(invoice_name: str, invoice_text: str, owner = None) -> ReportID:
     if not invoice_name.endswith('.xml'):
         raise InputError(detail="Invoice file type is not XML")
 
-    return {
-        "report_id": generate_report(invoice_name, invoice_text)
-    }
+    return ReportID(report_id=generate_report(invoice_name, invoice_text, owner))
 
 def invoice_check_validity_v1(report_id: int) -> CheckValidReturn:
     if report_id < 0:
@@ -55,7 +47,7 @@ def invoice_check_validity_v1(report_id: int) -> CheckValidReturn:
     return CheckValidReturn(is_valid=report.is_valid)
 
 def invoice_generate_hash_v1(invoice: TextInvoice) -> str:
-    return {}
+    return ""
 
-def invoice_upload_bulk_text_v1(invoices: List[TextInvoice]) -> ReportIDs:
-    return ReportIDs(report_ids=[generate_report(invoice.name, invoice.text) for invoice in invoices])
+def invoice_upload_bulk_text_v1(invoices: List[TextInvoice], owner = None) -> ReportIDs:
+    return ReportIDs(report_ids=[generate_report(invoice.name, invoice.text, owner) for invoice in invoices])