From 45ffaf4fc138115ee6244c2079bd3a5cbb9fb7eb Mon Sep 17 00:00:00 2001 From: emjay0921 Date: Thu, 19 Mar 2026 11:17:23 +0800 Subject: [PATCH 01/18] feat: migrate 14 modules from private to public repo Modules moved: - spp_land_record - spp_registry_group_hierarchy - spp_farmer_registry - spp_farmer_registry_cr - spp_farmer_registry_demo - spp_farmer_registry_vocabularies - spp_starter_farmer_registry - spp_farmer_registry_dashboard - spp_scoring - spp_scoring_programs - spp_disability_registry - spp_irrigation - spp_attachment_av_scan - spp_storage_backend --- spp_attachment_av_scan/README.md | 211 ++ spp_attachment_av_scan/README.rst | 178 ++ spp_attachment_av_scan/__init__.py | 1 + spp_attachment_av_scan/__manifest__.py | 35 + .../data/av_scanner_data.xml | 15 + .../data/quarantine_cron.xml | 37 + spp_attachment_av_scan/models/__init__.py | 2 + .../models/av_scanner_backend.py | 326 +++ .../models/ir_attachment.py | 765 ++++++ spp_attachment_av_scan/pyproject.toml | 3 + spp_attachment_av_scan/readme/DESCRIPTION.md | 65 + .../security/ir.model.access.csv | 3 + spp_attachment_av_scan/security/security.xml | 44 + .../static/description/icon.png | Bin 0 -> 15480 bytes .../static/description/index.html | 551 ++++ spp_attachment_av_scan/tests/__init__.py | 2 + .../tests/test_av_scanner_backend.py | 281 ++ .../tests/test_ir_attachment.py | 730 ++++++ .../views/av_scanner_backend_views.xml | 106 + .../views/ir_attachment_views.xml | 145 ++ spp_disability_registry/__init__.py | 11 + spp_disability_registry/__manifest__.py | 44 + .../data/concept_groups.xml | 45 + spp_disability_registry/data/id_type.xml | 8 + .../data/vocabulary_cause.xml | 105 + .../data/vocabulary_device.xml | 195 ++ .../data/vocabulary_severity.xml | 51 + .../data/vocabulary_type.xml | 150 ++ spp_disability_registry/demo/demo.xml | 102 + spp_disability_registry/models/__init__.py | 4 + spp_disability_registry/models/assessment.py | 291 +++ .../models/assistive_device.py | 79 + .../models/cel_disability_functions.py | 108 + spp_disability_registry/models/registrant.py | 129 + spp_disability_registry/pyproject.toml | 3 + spp_disability_registry/security/groups.xml | 60 + .../security/ir.model.access.csv | 9 + .../security/record_rules.xml | 62 + spp_disability_registry/services/__init__.py | 1 + .../services/cel_functions.py | 172 ++ .../static/description/icon.png | Bin 0 -> 15480 bytes spp_disability_registry/tests/__init__.py | 3 + .../tests/test_assessment.py | 389 +++ .../tests/test_cel_functions.py | 361 +++ .../tests/test_registrant.py | 472 ++++ .../views/assessment_views.xml | 358 +++ .../views/assistive_device_views.xml | 145 ++ spp_disability_registry/views/menus.xml | 38 + .../views/registrant_views.xml | 268 ++ spp_farmer_registry/__init__.py | 45 + spp_farmer_registry/__manifest__.py | 73 + spp_farmer_registry/controllers/__init__.py | 2 + spp_farmer_registry/controllers/main.py | 15 + spp_farmer_registry/data/cel_constants.xml | 183 ++ .../data/cel_variable_categories.xml | 33 + spp_farmer_registry/data/cel_variables.xml | 223 ++ .../data/config_parameters.xml | 8 + spp_farmer_registry/data/user_roles.xml | 38 + spp_farmer_registry/models/__init__.py | 8 + spp_farmer_registry/models/cel_variable.py | 77 + spp_farmer_registry/models/farm.py | 395 +++ spp_farmer_registry/models/farm_activity.py | 256 ++ spp_farmer_registry/models/farm_asset.py | 90 + spp_farmer_registry/models/farm_details.py | 124 + spp_farmer_registry/models/farm_season.py | 300 +++ .../models/res_config_settings.py | 17 + spp_farmer_registry/pyproject.toml | 3 + .../security/ir.model.access.csv | 40 + spp_farmer_registry/security/privileges.xml | 10 + spp_farmer_registry/security/security.xml | 101 + .../static/description/icon.png | Bin 0 -> 15480 bytes .../static/src/js/registry_restriction.js | 271 ++ spp_farmer_registry/tests/__init__.py | 9 + spp_farmer_registry/tests/common.py | 401 +++ .../tests/test_cel_variable.py | 320 +++ spp_farmer_registry/tests/test_controller.py | 59 + spp_farmer_registry/tests/test_farm.py | 643 +++++ .../tests/test_farm_activity.py | 312 +++ spp_farmer_registry/tests/test_farm_asset.py | 199 ++ .../tests/test_farm_details.py | 192 ++ spp_farmer_registry/tests/test_farm_season.py | 524 ++++ .../tests/test_res_config_settings.py | 66 + .../views/farm_activity_views.xml | 133 + .../views/farm_details_views.xml | 58 + .../views/farm_season_views.xml | 124 + spp_farmer_registry/views/farm_views.xml | 174 ++ .../views/res_config_settings_views.xml | 26 + spp_farmer_registry_cr/__init__.py | 5 + spp_farmer_registry_cr/__manifest__.py | 49 + .../data/approval_definitions.xml | 49 + spp_farmer_registry_cr/data/cr_types.xml | 155 ++ spp_farmer_registry_cr/details/__init__.py | 5 + .../details/cr_farm_activity.py | 242 ++ .../details/cr_farm_asset.py | 173 ++ .../details/cr_farm_details.py | 95 + .../details/cr_land_parcel.py | 148 ++ spp_farmer_registry_cr/models/__init__.py | 2 + .../models/change_request_type.py | 70 + spp_farmer_registry_cr/pyproject.toml | 3 + .../security/ir.model.access.csv | 13 + .../static/description/icon.png | Bin 0 -> 15480 bytes spp_farmer_registry_cr/strategies/__init__.py | 5 + .../strategies/manage_farm_activity.py | 256 ++ .../strategies/manage_farm_asset.py | 201 ++ .../strategies/manage_land_parcel.py | 212 ++ .../strategies/update_farm_details.py | 129 + spp_farmer_registry_cr/tests/__init__.py | 9 + spp_farmer_registry_cr/tests/common.py | 478 ++++ .../tests/test_change_request_type.py | 111 + .../tests/test_cr_farm_activity.py | 732 ++++++ .../tests/test_cr_farm_details.py | 408 +++ .../tests/test_strategy_farm_details.py | 368 +++ .../tests/test_strategy_manage_activity.py | 721 ++++++ .../views/change_request_type_views.xml | 71 + .../views/cr_farm_activity_views.xml | 229 ++ .../views/cr_farm_asset_views.xml | 131 + .../views/cr_farm_details_views.xml | 103 + .../views/cr_land_parcel_views.xml | 117 + spp_farmer_registry_dashboard/__init__.py | 1 + spp_farmer_registry_dashboard/__manifest__.py | 46 + .../data/dashboard_metrics.xml | 182 ++ .../data/dashboards.xml | 29 + .../data/files/farmer_overview.json | 539 ++++ spp_farmer_registry_dashboard/pyproject.toml | 3 + .../static/description/icon.png | Bin 0 -> 15480 bytes spp_farmer_registry_demo/__init__.py | 2 + spp_farmer_registry_demo/__manifest__.py | 70 + .../data/approval_links.xml | 52 + .../data/demo_personas.xml | 387 +++ .../data/demo_programs.xml | 43 + spp_farmer_registry_demo/data/demo_users.xml | 119 + spp_farmer_registry_demo/data/logic_packs.xml | 239 ++ spp_farmer_registry_demo/docs/USE_CASES.md | 894 +++++++ spp_farmer_registry_demo/models/__init__.py | 5 + .../models/demo_programs.py | 306 +++ .../models/farmer_blueprints.py | 632 +++++ .../models/farmer_demo_generator.py | 2268 +++++++++++++++++ .../models/seeded_farm_generator.py | 865 +++++++ spp_farmer_registry_demo/pyproject.toml | 3 + .../security/ir.model.access.csv | 2 + .../static/description/icon.png | Bin 0 -> 15480 bytes spp_farmer_registry_demo/tests/__init__.py | 5 + .../tests/test_blueprint_reproducibility.py | 423 +++ .../tests/test_demo_generator.py | 563 ++++ .../tests/test_seeded_farm_generator.py | 1088 ++++++++ .../tests/test_story_change_requests.py | 541 ++++ .../views/farmer_demo_wizard_view.xml | 81 + spp_farmer_registry_vocabularies/README.md | 202 ++ spp_farmer_registry_vocabularies/__init__.py | 3 + .../__manifest__.py | 39 + .../data/vocab_activity_purpose.xml | 34 + .../data/vocab_aquaculture_default.xml | 198 ++ .../data/vocab_crops_default.xml | 311 +++ .../data/vocab_cultivation_method.xml | 26 + .../data/vocab_data_source.xml | 42 + .../data/vocab_farm_type.xml | 42 + .../data/vocab_holder_type.xml | 36 + .../data/vocab_irrigation_asset_type.xml | 66 + .../data/vocab_land_tenure.xml | 66 + .../data/vocab_land_use.xml | 66 + .../data/vocab_livestock_default.xml | 210 ++ .../models/__init__.py | 2 + .../models/agrovoc_import.py | 558 ++++ .../pyproject.toml | 3 + .../security/ir.model.access.csv | 3 + .../static/description/icon.png | Bin 0 -> 15480 bytes .../tests/__init__.py | 3 + .../tests/test_agrovoc_import.py | 485 ++++ .../tests/test_agrovoc_wizard.py | 247 ++ .../wizard/__init__.py | 2 + .../wizard/agrovoc_import_wizard.py | 192 ++ .../wizard/agrovoc_import_wizard_view.xml | 182 ++ spp_irrigation/README.rst | 140 + spp_irrigation/__init__.py | 2 + spp_irrigation/__manifest__.py | 35 + spp_irrigation/i18n/lo.po | 120 + spp_irrigation/i18n/spp_irrigation.pot | 119 + spp_irrigation/models/__init__.py | 1 + spp_irrigation/models/irrigation.py | 61 + spp_irrigation/pyproject.toml | 3 + spp_irrigation/readme/DESCRIPTION.md | 44 + spp_irrigation/security/ir.model.access.csv | 3 + .../security/irrigation_security.xml | 15 + spp_irrigation/security/privileges.xml | 9 + spp_irrigation/static/description/icon.png | Bin 0 -> 15480 bytes spp_irrigation/static/description/index.html | 498 ++++ spp_irrigation/tests/__init__.py | 3 + spp_irrigation/tests/test_irrigation_asset.py | 194 ++ .../tests/test_irrigation_edge_cases.py | 526 ++++ spp_irrigation/views/irrigation_view.xml | 86 + spp_land_record/README.rst | 132 + spp_land_record/__init__.py | 3 + spp_land_record/__manifest__.py | 34 + spp_land_record/i18n/lo.po | 146 ++ spp_land_record/i18n/spp_land_record.pot | 134 + spp_land_record/models/__init__.py | 3 + spp_land_record/models/land_record.py | 157 ++ spp_land_record/pyproject.toml | 3 + spp_land_record/readme/DESCRIPTION.md | 39 + spp_land_record/security/ir.model.access.csv | 7 + spp_land_record/static/description/icon.png | Bin 0 -> 15480 bytes spp_land_record/static/description/index.html | 499 ++++ spp_land_record/tests/__init__.py | 2 + spp_land_record/tests/test_land_record.py | 126 + .../tests/test_land_record_validation.py | 578 +++++ spp_land_record/views/land_record_views.xml | 140 + spp_registry_group_hierarchy/README.rst | 131 + spp_registry_group_hierarchy/__init__.py | 1 + spp_registry_group_hierarchy/__manifest__.py | 34 + spp_registry_group_hierarchy/i18n/lo.po | 93 + .../i18n/spp_registry_group_hierarchy.pot | 85 + .../models/__init__.py | 2 + .../models/group_membership.py | 54 + .../models/vocabulary_code.py | 21 + spp_registry_group_hierarchy/pyproject.toml | 3 + .../readme/DESCRIPTION.md | 37 + .../security/ir.model.access.csv | 1 + .../static/description/icon.png | Bin 0 -> 15480 bytes .../static/description/index.html | 483 ++++ .../tests/__init__.py | 1 + .../tests/test_group_membership.py | 133 + .../views/group_membership_views.xml | 40 + .../views/group_views.xml | 83 + .../views/vocabulary_code_views.xml | 14 + spp_scoring/README.rst | 179 ++ spp_scoring/__init__.py | 28 + spp_scoring/__manifest__.py | 66 + spp_scoring/data/job_worker_channel.xml | 12 + spp_scoring/i18n/fr.po | 1930 ++++++++++++++ spp_scoring/i18n/spp_scoring.pot | 1887 ++++++++++++++ spp_scoring/models/__init__.py | 10 + spp_scoring/models/scoring_batch_job.py | 75 + .../models/scoring_data_integration.py | 205 ++ spp_scoring/models/scoring_engine.py | 598 +++++ spp_scoring/models/scoring_indicator.py | 392 +++ .../models/scoring_indicator_provider.py | 191 ++ spp_scoring/models/scoring_model.py | 300 +++ spp_scoring/models/scoring_result.py | 243 ++ spp_scoring/models/scoring_result_detail.py | 65 + spp_scoring/models/scoring_threshold.py | 108 + spp_scoring/models/scoring_value_mapping.py | 72 + spp_scoring/pyproject.toml | 3 + spp_scoring/readme/DESCRIPTION.md | 62 + spp_scoring/security/compliance.yaml | 290 +++ spp_scoring/security/groups.xml | 76 + spp_scoring/security/ir.model.access.csv | 24 + spp_scoring/security/privileges.xml | 14 + spp_scoring/security/rules.xml | 84 + spp_scoring/static/description/icon.png | Bin 0 -> 15480 bytes spp_scoring/static/description/index.html | 553 ++++ spp_scoring/tests/__init__.py | 8 + .../tests/test_batch_scoring_wizard.py | 261 ++ spp_scoring/tests/test_data_integrity.py | 377 +++ spp_scoring/tests/test_edge_cases.py | 496 ++++ spp_scoring/tests/test_scoring_engine.py | 338 +++ spp_scoring/tests/test_scoring_indicator.py | 252 ++ .../tests/test_scoring_indicator_provider.py | 305 +++ spp_scoring/tests/test_scoring_model.py | 214 ++ spp_scoring/tests/test_scoring_workflow.py | 337 +++ spp_scoring/views/menus.xml | 65 + spp_scoring/views/scoring_indicator_views.xml | 115 + spp_scoring/views/scoring_model_views.xml | 173 ++ spp_scoring/views/scoring_result_views.xml | 173 ++ spp_scoring/views/scoring_threshold_views.xml | 89 + spp_scoring/wizard/__init__.py | 1 + spp_scoring/wizard/batch_scoring_wizard.py | 148 ++ .../wizard/batch_scoring_wizard_views.xml | 60 + spp_scoring_programs/README.rst | 160 ++ spp_scoring_programs/__init__.py | 1 + spp_scoring_programs/__manifest__.py | 39 + spp_scoring_programs/models/__init__.py | 2 + spp_scoring_programs/models/program.py | 202 ++ spp_scoring_programs/models/scoring_model.py | 35 + spp_scoring_programs/pyproject.toml | 3 + spp_scoring_programs/readme/DESCRIPTION.md | 54 + .../static/description/icon.png | Bin 0 -> 15480 bytes .../static/description/index.html | 510 ++++ spp_scoring_programs/tests/__init__.py | 1 + .../tests/test_scoring_programs.py | 258 ++ spp_scoring_programs/views/program_views.xml | 70 + .../views/scoring_model_views.xml | 43 + spp_starter_farmer_registry/__init__.py | 1 + spp_starter_farmer_registry/__manifest__.py | 77 + .../data/config_parameters.xml | 23 + spp_starter_farmer_registry/pyproject.toml | 3 + .../static/description/icon.png | Bin 0 -> 15480 bytes spp_storage_backend/README.md | 223 ++ spp_storage_backend/README.rst | 152 ++ spp_storage_backend/__init__.py | 3 + spp_storage_backend/__manifest__.py | 30 + .../data/storage_backend_data.xml | 13 + spp_storage_backend/models/__init__.py | 3 + spp_storage_backend/models/storage_backend.py | 603 +++++ spp_storage_backend/pyproject.toml | 3 + spp_storage_backend/readme/DESCRIPTION.md | 53 + spp_storage_backend/security/groups.xml | 23 + .../security/ir.model.access.csv | 3 + spp_storage_backend/security/privileges.xml | 16 + .../static/description/icon.png | Bin 0 -> 15480 bytes .../static/description/index.html | 517 ++++ spp_storage_backend/tests/__init__.py | 3 + .../tests/test_storage_backend.py | 473 ++++ .../views/storage_backend_views.xml | 128 + 303 files changed, 50096 insertions(+) create mode 100644 spp_attachment_av_scan/README.md create mode 100644 spp_attachment_av_scan/README.rst create mode 100644 spp_attachment_av_scan/__init__.py create mode 100644 spp_attachment_av_scan/__manifest__.py create mode 100644 spp_attachment_av_scan/data/av_scanner_data.xml create mode 100644 spp_attachment_av_scan/data/quarantine_cron.xml create mode 100644 spp_attachment_av_scan/models/__init__.py create mode 100644 spp_attachment_av_scan/models/av_scanner_backend.py create mode 100644 spp_attachment_av_scan/models/ir_attachment.py create mode 100644 spp_attachment_av_scan/pyproject.toml create mode 100644 spp_attachment_av_scan/readme/DESCRIPTION.md create mode 100644 spp_attachment_av_scan/security/ir.model.access.csv create mode 100644 spp_attachment_av_scan/security/security.xml create mode 100644 spp_attachment_av_scan/static/description/icon.png create mode 100644 spp_attachment_av_scan/static/description/index.html create mode 100644 spp_attachment_av_scan/tests/__init__.py create mode 100644 spp_attachment_av_scan/tests/test_av_scanner_backend.py create mode 100644 spp_attachment_av_scan/tests/test_ir_attachment.py create mode 100644 spp_attachment_av_scan/views/av_scanner_backend_views.xml create mode 100644 spp_attachment_av_scan/views/ir_attachment_views.xml create mode 100644 spp_disability_registry/__init__.py create mode 100644 spp_disability_registry/__manifest__.py create mode 100644 spp_disability_registry/data/concept_groups.xml create mode 100644 spp_disability_registry/data/id_type.xml create mode 100644 spp_disability_registry/data/vocabulary_cause.xml create mode 100644 spp_disability_registry/data/vocabulary_device.xml create mode 100644 spp_disability_registry/data/vocabulary_severity.xml create mode 100644 spp_disability_registry/data/vocabulary_type.xml create mode 100644 spp_disability_registry/demo/demo.xml create mode 100644 spp_disability_registry/models/__init__.py create mode 100644 spp_disability_registry/models/assessment.py create mode 100644 spp_disability_registry/models/assistive_device.py create mode 100644 spp_disability_registry/models/cel_disability_functions.py create mode 100644 spp_disability_registry/models/registrant.py create mode 100644 spp_disability_registry/pyproject.toml create mode 100644 spp_disability_registry/security/groups.xml create mode 100644 spp_disability_registry/security/ir.model.access.csv create mode 100644 spp_disability_registry/security/record_rules.xml create mode 100644 spp_disability_registry/services/__init__.py create mode 100644 spp_disability_registry/services/cel_functions.py create mode 100644 spp_disability_registry/static/description/icon.png create mode 100644 spp_disability_registry/tests/__init__.py create mode 100644 spp_disability_registry/tests/test_assessment.py create mode 100644 spp_disability_registry/tests/test_cel_functions.py create mode 100644 spp_disability_registry/tests/test_registrant.py create mode 100644 spp_disability_registry/views/assessment_views.xml create mode 100644 spp_disability_registry/views/assistive_device_views.xml create mode 100644 spp_disability_registry/views/menus.xml create mode 100644 spp_disability_registry/views/registrant_views.xml create mode 100644 spp_farmer_registry/__init__.py create mode 100644 spp_farmer_registry/__manifest__.py create mode 100644 spp_farmer_registry/controllers/__init__.py create mode 100644 spp_farmer_registry/controllers/main.py create mode 100644 spp_farmer_registry/data/cel_constants.xml create mode 100644 spp_farmer_registry/data/cel_variable_categories.xml create mode 100644 spp_farmer_registry/data/cel_variables.xml create mode 100644 spp_farmer_registry/data/config_parameters.xml create mode 100644 spp_farmer_registry/data/user_roles.xml create mode 100644 spp_farmer_registry/models/__init__.py create mode 100644 spp_farmer_registry/models/cel_variable.py create mode 100644 spp_farmer_registry/models/farm.py create mode 100644 spp_farmer_registry/models/farm_activity.py create mode 100644 spp_farmer_registry/models/farm_asset.py create mode 100644 spp_farmer_registry/models/farm_details.py create mode 100644 spp_farmer_registry/models/farm_season.py create mode 100644 spp_farmer_registry/models/res_config_settings.py create mode 100644 spp_farmer_registry/pyproject.toml create mode 100644 spp_farmer_registry/security/ir.model.access.csv create mode 100644 spp_farmer_registry/security/privileges.xml create mode 100644 spp_farmer_registry/security/security.xml create mode 100644 spp_farmer_registry/static/description/icon.png create mode 100644 spp_farmer_registry/static/src/js/registry_restriction.js create mode 100644 spp_farmer_registry/tests/__init__.py create mode 100644 spp_farmer_registry/tests/common.py create mode 100644 spp_farmer_registry/tests/test_cel_variable.py create mode 100644 spp_farmer_registry/tests/test_controller.py create mode 100644 spp_farmer_registry/tests/test_farm.py create mode 100644 spp_farmer_registry/tests/test_farm_activity.py create mode 100644 spp_farmer_registry/tests/test_farm_asset.py create mode 100644 spp_farmer_registry/tests/test_farm_details.py create mode 100644 spp_farmer_registry/tests/test_farm_season.py create mode 100644 spp_farmer_registry/tests/test_res_config_settings.py create mode 100644 spp_farmer_registry/views/farm_activity_views.xml create mode 100644 spp_farmer_registry/views/farm_details_views.xml create mode 100644 spp_farmer_registry/views/farm_season_views.xml create mode 100644 spp_farmer_registry/views/farm_views.xml create mode 100644 spp_farmer_registry/views/res_config_settings_views.xml create mode 100644 spp_farmer_registry_cr/__init__.py create mode 100644 spp_farmer_registry_cr/__manifest__.py create mode 100644 spp_farmer_registry_cr/data/approval_definitions.xml create mode 100644 spp_farmer_registry_cr/data/cr_types.xml create mode 100644 spp_farmer_registry_cr/details/__init__.py create mode 100644 spp_farmer_registry_cr/details/cr_farm_activity.py create mode 100644 spp_farmer_registry_cr/details/cr_farm_asset.py create mode 100644 spp_farmer_registry_cr/details/cr_farm_details.py create mode 100644 spp_farmer_registry_cr/details/cr_land_parcel.py create mode 100644 spp_farmer_registry_cr/models/__init__.py create mode 100644 spp_farmer_registry_cr/models/change_request_type.py create mode 100644 spp_farmer_registry_cr/pyproject.toml create mode 100644 spp_farmer_registry_cr/security/ir.model.access.csv create mode 100644 spp_farmer_registry_cr/static/description/icon.png create mode 100644 spp_farmer_registry_cr/strategies/__init__.py create mode 100644 spp_farmer_registry_cr/strategies/manage_farm_activity.py create mode 100644 spp_farmer_registry_cr/strategies/manage_farm_asset.py create mode 100644 spp_farmer_registry_cr/strategies/manage_land_parcel.py create mode 100644 spp_farmer_registry_cr/strategies/update_farm_details.py create mode 100644 spp_farmer_registry_cr/tests/__init__.py create mode 100644 spp_farmer_registry_cr/tests/common.py create mode 100644 spp_farmer_registry_cr/tests/test_change_request_type.py create mode 100644 spp_farmer_registry_cr/tests/test_cr_farm_activity.py create mode 100644 spp_farmer_registry_cr/tests/test_cr_farm_details.py create mode 100644 spp_farmer_registry_cr/tests/test_strategy_farm_details.py create mode 100644 spp_farmer_registry_cr/tests/test_strategy_manage_activity.py create mode 100644 spp_farmer_registry_cr/views/change_request_type_views.xml create mode 100644 spp_farmer_registry_cr/views/cr_farm_activity_views.xml create mode 100644 spp_farmer_registry_cr/views/cr_farm_asset_views.xml create mode 100644 spp_farmer_registry_cr/views/cr_farm_details_views.xml create mode 100644 spp_farmer_registry_cr/views/cr_land_parcel_views.xml create mode 100644 spp_farmer_registry_dashboard/__init__.py create mode 100644 spp_farmer_registry_dashboard/__manifest__.py create mode 100644 spp_farmer_registry_dashboard/data/dashboard_metrics.xml create mode 100644 spp_farmer_registry_dashboard/data/dashboards.xml create mode 100644 spp_farmer_registry_dashboard/data/files/farmer_overview.json create mode 100644 spp_farmer_registry_dashboard/pyproject.toml create mode 100644 spp_farmer_registry_dashboard/static/description/icon.png create mode 100644 spp_farmer_registry_demo/__init__.py create mode 100644 spp_farmer_registry_demo/__manifest__.py create mode 100644 spp_farmer_registry_demo/data/approval_links.xml create mode 100644 spp_farmer_registry_demo/data/demo_personas.xml create mode 100644 spp_farmer_registry_demo/data/demo_programs.xml create mode 100644 spp_farmer_registry_demo/data/demo_users.xml create mode 100644 spp_farmer_registry_demo/data/logic_packs.xml create mode 100644 spp_farmer_registry_demo/docs/USE_CASES.md create mode 100644 spp_farmer_registry_demo/models/__init__.py create mode 100644 spp_farmer_registry_demo/models/demo_programs.py create mode 100644 spp_farmer_registry_demo/models/farmer_blueprints.py create mode 100644 spp_farmer_registry_demo/models/farmer_demo_generator.py create mode 100644 spp_farmer_registry_demo/models/seeded_farm_generator.py create mode 100644 spp_farmer_registry_demo/pyproject.toml create mode 100644 spp_farmer_registry_demo/security/ir.model.access.csv create mode 100644 spp_farmer_registry_demo/static/description/icon.png create mode 100644 spp_farmer_registry_demo/tests/__init__.py create mode 100644 spp_farmer_registry_demo/tests/test_blueprint_reproducibility.py create mode 100644 spp_farmer_registry_demo/tests/test_demo_generator.py create mode 100644 spp_farmer_registry_demo/tests/test_seeded_farm_generator.py create mode 100644 spp_farmer_registry_demo/tests/test_story_change_requests.py create mode 100644 spp_farmer_registry_demo/views/farmer_demo_wizard_view.xml create mode 100644 spp_farmer_registry_vocabularies/README.md create mode 100644 spp_farmer_registry_vocabularies/__init__.py create mode 100644 spp_farmer_registry_vocabularies/__manifest__.py create mode 100644 spp_farmer_registry_vocabularies/data/vocab_activity_purpose.xml create mode 100644 spp_farmer_registry_vocabularies/data/vocab_aquaculture_default.xml create mode 100644 spp_farmer_registry_vocabularies/data/vocab_crops_default.xml create mode 100644 spp_farmer_registry_vocabularies/data/vocab_cultivation_method.xml create mode 100644 spp_farmer_registry_vocabularies/data/vocab_data_source.xml create mode 100644 spp_farmer_registry_vocabularies/data/vocab_farm_type.xml create mode 100644 spp_farmer_registry_vocabularies/data/vocab_holder_type.xml create mode 100644 spp_farmer_registry_vocabularies/data/vocab_irrigation_asset_type.xml create mode 100644 spp_farmer_registry_vocabularies/data/vocab_land_tenure.xml create mode 100644 spp_farmer_registry_vocabularies/data/vocab_land_use.xml create mode 100644 spp_farmer_registry_vocabularies/data/vocab_livestock_default.xml create mode 100644 spp_farmer_registry_vocabularies/models/__init__.py create mode 100644 spp_farmer_registry_vocabularies/models/agrovoc_import.py create mode 100644 spp_farmer_registry_vocabularies/pyproject.toml create mode 100644 spp_farmer_registry_vocabularies/security/ir.model.access.csv create mode 100644 spp_farmer_registry_vocabularies/static/description/icon.png create mode 100644 spp_farmer_registry_vocabularies/tests/__init__.py create mode 100644 spp_farmer_registry_vocabularies/tests/test_agrovoc_import.py create mode 100644 spp_farmer_registry_vocabularies/tests/test_agrovoc_wizard.py create mode 100644 spp_farmer_registry_vocabularies/wizard/__init__.py create mode 100644 spp_farmer_registry_vocabularies/wizard/agrovoc_import_wizard.py create mode 100644 spp_farmer_registry_vocabularies/wizard/agrovoc_import_wizard_view.xml create mode 100644 spp_irrigation/README.rst create mode 100644 spp_irrigation/__init__.py create mode 100644 spp_irrigation/__manifest__.py create mode 100644 spp_irrigation/i18n/lo.po create mode 100644 spp_irrigation/i18n/spp_irrigation.pot create mode 100644 spp_irrigation/models/__init__.py create mode 100644 spp_irrigation/models/irrigation.py create mode 100644 spp_irrigation/pyproject.toml create mode 100644 spp_irrigation/readme/DESCRIPTION.md create mode 100644 spp_irrigation/security/ir.model.access.csv create mode 100644 spp_irrigation/security/irrigation_security.xml create mode 100644 spp_irrigation/security/privileges.xml create mode 100644 spp_irrigation/static/description/icon.png create mode 100644 spp_irrigation/static/description/index.html create mode 100644 spp_irrigation/tests/__init__.py create mode 100644 spp_irrigation/tests/test_irrigation_asset.py create mode 100644 spp_irrigation/tests/test_irrigation_edge_cases.py create mode 100644 spp_irrigation/views/irrigation_view.xml create mode 100644 spp_land_record/README.rst create mode 100644 spp_land_record/__init__.py create mode 100644 spp_land_record/__manifest__.py create mode 100644 spp_land_record/i18n/lo.po create mode 100644 spp_land_record/i18n/spp_land_record.pot create mode 100644 spp_land_record/models/__init__.py create mode 100644 spp_land_record/models/land_record.py create mode 100644 spp_land_record/pyproject.toml create mode 100644 spp_land_record/readme/DESCRIPTION.md create mode 100644 spp_land_record/security/ir.model.access.csv create mode 100644 spp_land_record/static/description/icon.png create mode 100644 spp_land_record/static/description/index.html create mode 100644 spp_land_record/tests/__init__.py create mode 100644 spp_land_record/tests/test_land_record.py create mode 100644 spp_land_record/tests/test_land_record_validation.py create mode 100644 spp_land_record/views/land_record_views.xml create mode 100644 spp_registry_group_hierarchy/README.rst create mode 100644 spp_registry_group_hierarchy/__init__.py create mode 100644 spp_registry_group_hierarchy/__manifest__.py create mode 100644 spp_registry_group_hierarchy/i18n/lo.po create mode 100644 spp_registry_group_hierarchy/i18n/spp_registry_group_hierarchy.pot create mode 100644 spp_registry_group_hierarchy/models/__init__.py create mode 100644 spp_registry_group_hierarchy/models/group_membership.py create mode 100644 spp_registry_group_hierarchy/models/vocabulary_code.py create mode 100644 spp_registry_group_hierarchy/pyproject.toml create mode 100644 spp_registry_group_hierarchy/readme/DESCRIPTION.md create mode 100644 spp_registry_group_hierarchy/security/ir.model.access.csv create mode 100644 spp_registry_group_hierarchy/static/description/icon.png create mode 100644 spp_registry_group_hierarchy/static/description/index.html create mode 100644 spp_registry_group_hierarchy/tests/__init__.py create mode 100644 spp_registry_group_hierarchy/tests/test_group_membership.py create mode 100644 spp_registry_group_hierarchy/views/group_membership_views.xml create mode 100644 spp_registry_group_hierarchy/views/group_views.xml create mode 100644 spp_registry_group_hierarchy/views/vocabulary_code_views.xml create mode 100644 spp_scoring/README.rst create mode 100644 spp_scoring/__init__.py create mode 100644 spp_scoring/__manifest__.py create mode 100644 spp_scoring/data/job_worker_channel.xml create mode 100644 spp_scoring/i18n/fr.po create mode 100644 spp_scoring/i18n/spp_scoring.pot create mode 100644 spp_scoring/models/__init__.py create mode 100644 spp_scoring/models/scoring_batch_job.py create mode 100644 spp_scoring/models/scoring_data_integration.py create mode 100644 spp_scoring/models/scoring_engine.py create mode 100644 spp_scoring/models/scoring_indicator.py create mode 100644 spp_scoring/models/scoring_indicator_provider.py create mode 100644 spp_scoring/models/scoring_model.py create mode 100644 spp_scoring/models/scoring_result.py create mode 100644 spp_scoring/models/scoring_result_detail.py create mode 100644 spp_scoring/models/scoring_threshold.py create mode 100644 spp_scoring/models/scoring_value_mapping.py create mode 100644 spp_scoring/pyproject.toml create mode 100644 spp_scoring/readme/DESCRIPTION.md create mode 100644 spp_scoring/security/compliance.yaml create mode 100644 spp_scoring/security/groups.xml create mode 100644 spp_scoring/security/ir.model.access.csv create mode 100644 spp_scoring/security/privileges.xml create mode 100644 spp_scoring/security/rules.xml create mode 100644 spp_scoring/static/description/icon.png create mode 100644 spp_scoring/static/description/index.html create mode 100644 spp_scoring/tests/__init__.py create mode 100644 spp_scoring/tests/test_batch_scoring_wizard.py create mode 100644 spp_scoring/tests/test_data_integrity.py create mode 100644 spp_scoring/tests/test_edge_cases.py create mode 100644 spp_scoring/tests/test_scoring_engine.py create mode 100644 spp_scoring/tests/test_scoring_indicator.py create mode 100644 spp_scoring/tests/test_scoring_indicator_provider.py create mode 100644 spp_scoring/tests/test_scoring_model.py create mode 100644 spp_scoring/tests/test_scoring_workflow.py create mode 100644 spp_scoring/views/menus.xml create mode 100644 spp_scoring/views/scoring_indicator_views.xml create mode 100644 spp_scoring/views/scoring_model_views.xml create mode 100644 spp_scoring/views/scoring_result_views.xml create mode 100644 spp_scoring/views/scoring_threshold_views.xml create mode 100644 spp_scoring/wizard/__init__.py create mode 100644 spp_scoring/wizard/batch_scoring_wizard.py create mode 100644 spp_scoring/wizard/batch_scoring_wizard_views.xml create mode 100644 spp_scoring_programs/README.rst create mode 100644 spp_scoring_programs/__init__.py create mode 100644 spp_scoring_programs/__manifest__.py create mode 100644 spp_scoring_programs/models/__init__.py create mode 100644 spp_scoring_programs/models/program.py create mode 100644 spp_scoring_programs/models/scoring_model.py create mode 100644 spp_scoring_programs/pyproject.toml create mode 100644 spp_scoring_programs/readme/DESCRIPTION.md create mode 100644 spp_scoring_programs/static/description/icon.png create mode 100644 spp_scoring_programs/static/description/index.html create mode 100644 spp_scoring_programs/tests/__init__.py create mode 100644 spp_scoring_programs/tests/test_scoring_programs.py create mode 100644 spp_scoring_programs/views/program_views.xml create mode 100644 spp_scoring_programs/views/scoring_model_views.xml create mode 100644 spp_starter_farmer_registry/__init__.py create mode 100644 spp_starter_farmer_registry/__manifest__.py create mode 100644 spp_starter_farmer_registry/data/config_parameters.xml create mode 100644 spp_starter_farmer_registry/pyproject.toml create mode 100644 spp_starter_farmer_registry/static/description/icon.png create mode 100644 spp_storage_backend/README.md create mode 100644 spp_storage_backend/README.rst create mode 100644 spp_storage_backend/__init__.py create mode 100644 spp_storage_backend/__manifest__.py create mode 100644 spp_storage_backend/data/storage_backend_data.xml create mode 100644 spp_storage_backend/models/__init__.py create mode 100644 spp_storage_backend/models/storage_backend.py create mode 100644 spp_storage_backend/pyproject.toml create mode 100644 spp_storage_backend/readme/DESCRIPTION.md create mode 100644 spp_storage_backend/security/groups.xml create mode 100644 spp_storage_backend/security/ir.model.access.csv create mode 100644 spp_storage_backend/security/privileges.xml create mode 100644 spp_storage_backend/static/description/icon.png create mode 100644 spp_storage_backend/static/description/index.html create mode 100644 spp_storage_backend/tests/__init__.py create mode 100644 spp_storage_backend/tests/test_storage_backend.py create mode 100644 spp_storage_backend/views/storage_backend_views.xml diff --git a/spp_attachment_av_scan/README.md b/spp_attachment_av_scan/README.md new file mode 100644 index 00000000..8314908c --- /dev/null +++ b/spp_attachment_av_scan/README.md @@ -0,0 +1,211 @@ +# OpenSPP Attachment Antivirus Scan + +System-wide antivirus scanning for file attachments in OpenSPP. + +## Overview + +This module provides automatic malware scanning for all file attachments uploaded to the system using ClamAV antivirus +engine. It integrates with the `queue_job` module for asynchronous scanning to avoid blocking file uploads. + +## Features + +- **Automatic Scanning**: All binary attachments are automatically queued for malware scanning upon upload +- **Configurable Backends**: Support for ClamAV via Unix socket or network connection +- **Quarantine**: Infected files are automatically quarantined and access is blocked +- **Security Notifications**: Security administrators are notified when malware is detected +- **Manual Rescans**: Administrators can manually trigger rescans of attachments +- **File Size Limits**: Configurable maximum file size to avoid scanning large files +- **Scan Timeouts**: Configurable timeout to prevent long-running scans + +## Dependencies + +- `base`: Odoo base module +- `queue_job`: For asynchronous job processing +- `spp_security`: OpenSPP security module for security groups +- `pyclamd`: Python library for ClamAV integration (external) + +## Installation + +1. Install ClamAV on your server: + + ```bash + # Ubuntu/Debian + sudo apt-get install clamav clamav-daemon + + # Start the daemon + sudo systemctl start clamav-daemon + sudo systemctl enable clamav-daemon + ``` + +2. Install the Python library: + + ```bash + pip install pyclamd + ``` + +3. Install the module in Odoo: + - Update the apps list + - Search for "OpenSPP Attachment Antivirus Scan" + - Click Install + +## Configuration + +### Scanner Backend Setup + +1. Go to **Settings > Technical > Antivirus Scanners** +2. Edit the "Default ClamAV Scanner" record +3. Configure the connection settings: + - **Backend Type**: Choose "ClamAV Unix Socket" or "ClamAV Network" + - **Socket Path**: Path to ClamAV socket (default: `/var/run/clamav/clamd.sock`) + - Or **Host/Port**: Network connection details + - **Max File Size**: Maximum file size to scan in MB (default: 100) + - **Scan Timeout**: Maximum time for scan in seconds (default: 60) +4. Enable the backend by toggling the "Active" button +5. Click "Test Connection" to verify the configuration + +### Security Groups + +- **Antivirus Administrator** (`group_av_admin`): Can manage scanner backends and view detailed scan results + +## Usage + +### Automatic Scanning + +When a user uploads a file: + +1. The attachment is created immediately +2. A scan job is queued in the background +3. The scan status shows "Pending Scan" +4. Once scanned, the status updates to: + - **Clean**: No malware detected + - **Infected**: Malware detected (file is quarantined) + - **Error**: Scan failed + - **Skipped**: File too large or no scanner configured + +### Viewing Scan Status + +Scan status is visible on the attachment form and list views: + +- Navigate to any attachment (e.g., in Documents or via Technical > Attachments) +- Check the "Antivirus Scan" section for scan status and details + +### Manual Rescans + +As an AV Administrator: + +1. Open an attachment +2. Click the "Rescan" button +3. The file is queued for scanning + +### Infected Files + +When malware is detected: + +1. The file is marked as "Infected" +2. The threat name is recorded +3. The file is quarantined (access to file data is blocked) +4. Security administrators receive an activity notification +5. The file cannot be downloaded until reviewed + +## Technical Details + +### Models + +#### `spp.av.scanner.backend` + +Stores configuration for antivirus scanner backends. + +**Key Fields**: + +- `backend_type`: Type of scanner (ClamAV socket or network) +- `is_active`: Whether this backend is active +- `clamd_socket_path`: Path to ClamAV Unix socket +- `clamd_host`, `clamd_port`: Network connection details +- `max_file_size_mb`: Maximum file size to scan +- `scan_timeout_seconds`: Scan timeout + +**Key Methods**: + +- `scan_binary(binary_data, filename)`: Scan binary data for malware +- `test_connection()`: Test connection to scanner +- `get_active_scanner()`: Get the active scanner backend + +#### `ir.attachment` (inherited) + +Extended with antivirus scan fields and logic. + +**New Fields**: + +- `scan_status`: Status of malware scan +- `scan_date`: When the file was scanned +- `scan_result`: Detailed scan result (JSON) +- `threat_name`: Name of detected threat +- `is_quarantined`: Whether file is quarantined + +**Key Methods**: + +- `_scan_for_malware()`: Async job to scan attachment +- `_quarantine()`: Quarantine infected file +- `_notify_security_admins()`: Notify admins of infection +- `action_rescan()`: Manually trigger rescan + +### Queue Jobs + +The module uses `queue_job` for asynchronous scanning: + +- Scans are queued when attachments are created or updated +- Priority 20 for automatic scans, priority 10 for manual rescans +- Jobs are named "Scan attachment {id} for malware" + +### Security + +- AV Administrators can manage scanner backends +- All users can view scan status on their attachments +- Quarantined files block access to binary data via `read()` override + +## Logging + +The module uses structured logging: + +- Info level: Scan results, connection tests +- Warning level: Malware detections, configuration issues +- Error level: Scan failures, connection errors + +No PII (personally identifiable information) is logged. + +## Performance Considerations + +- File scanning is asynchronous via queue_job +- Large files can be skipped via `max_file_size_mb` setting +- Scan timeouts prevent long-running operations +- Failed scans don't block file uploads + +## Limitations + +- Currently only supports ClamAV +- Requires ClamAV daemon to be running +- Files are scanned after upload (not during) +- Very large files may be skipped + +## Future Enhancements + +- Support for additional antivirus engines +- Real-time scanning before upload completion +- Scan result caching +- Scheduled rescans of all attachments +- Quarantine management interface +- Detailed scan statistics and reports + +## License + +LGPL-3 + +## Author + +OpenSPP.org + +## Maintainers + +- jeremi +- gonzalesedwin1123 +- reichie020212 diff --git a/spp_attachment_av_scan/README.rst b/spp_attachment_av_scan/README.rst new file mode 100644 index 00000000..357e813e --- /dev/null +++ b/spp_attachment_av_scan/README.rst @@ -0,0 +1,178 @@ +================================= +OpenSPP Attachment Antivirus Scan +================================= + +.. + !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! + !! This file is generated by oca-gen-addon-readme !! + !! changes will be overwritten. !! + !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! + !! source digest: sha256:f6237a54ba392825fb72fd61a8e94c10528b2bde807e13bf33beeac7f7b156b7 + !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! + +.. |badge1| image:: https://img.shields.io/badge/maturity-Beta-yellow.png + :target: https://odoo-community.org/page/development-status + :alt: Beta +.. |badge2| image:: https://img.shields.io/badge/license-LGPL--3-blue.png + :target: http://www.gnu.org/licenses/lgpl-3.0-standalone.html + :alt: License: LGPL-3 +.. |badge3| image:: https://img.shields.io/badge/github-OpenSPP%2Fopenspp--modules-lightgray.png?logo=github + :target: https://github.com/OpenSPP/openspp-modules/tree/19.0/spp_attachment_av_scan + :alt: OpenSPP/openspp-modules + +|badge1| |badge2| |badge3| + +Scans uploaded file attachments for malware using ClamAV antivirus +engine. Automatically queues scans for binary attachments on create or +update using queue_job. Quarantines infected files by encrypting them +with spp_encryption and removing original data. Provides forensic tools +for security administrators to restore false positives or download +quarantined files for analysis. + +Key Capabilities +~~~~~~~~~~~~~~~~ + +- Auto-scan binary attachments on upload or update via background jobs +- Quarantine infected files with encrypted backup and SHA256 hash + verification +- Block read access to quarantined attachment data +- Manual rescan, restore, forensic download, and permanent deletion of + quarantined files +- Notify security administrators when malware is detected +- Scheduled cleanup of old quarantined files and forensic downloads +- Support ClamAV via Unix socket or network connection + +Key Models +~~~~~~~~~~ + ++----------------------------+----------------------------------------+ +| Model | Description | ++============================+========================================+ +| ``spp.av.scanner.backend`` | Configures ClamAV connection | +| | (socket/network) and limits | ++----------------------------+----------------------------------------+ +| ``ir.attachment`` | Extended with scan status, threat | +| | name, and quarantine | ++----------------------------+----------------------------------------+ + +Configuration +~~~~~~~~~~~~~ + +After installing: + +1. Navigate to **Settings > Administration > Antivirus Scanners** +2. Create a scanner backend with ClamAV connection details (default: + ``/var/run/clamav/clamd.sock``) +3. Click **Test Connection** to verify ClamAV is running +4. Set **Active** to enable scanning +5. Configure system parameters: + + - ``spp_attachment_av_scan.quarantine_encryption_provider_id``: + Encryption provider for quarantine + - ``spp_attachment_av_scan.quarantine_retention_days``: Days before + purging quarantined files (default: 90) + - ``spp_attachment_av_scan.forensic_download_retention_hours``: Hours + before cleaning forensic downloads (default: 24) + +UI Location +~~~~~~~~~~~ + +- **Scanner Configuration**: Settings > Administration > Antivirus + Scanners +- **Quarantined Files**: Settings > Technical > Security > Quarantined + Files +- **Attachment Forms**: Scan status and quarantine actions appear in + "Antivirus Scan" section + +Tabs +~~~~ + +**Scanner Backend form** (``spp.av.scanner.backend``): + +- **Connection Settings**: Unix socket or network configuration for + ClamAV +- **Connection Status**: Last connection test results and error details + +Security +~~~~~~~~ + ++----------------------+----------------------+----------------------+ +| Group | Model | Access | ++======================+======================+======================+ +| ``base.group_user`` | ``spp. | Read | +| | av.scanner.backend`` | | ++----------------------+----------------------+----------------------+ +| ``base.group_user`` | ``ir.attachment`` | Read scan status | ++----------------------+----------------------+----------------------+ +| ` | ``spp. | Full CRUD | +| `spp_attachment_av_s | av.scanner.backend`` | | +| can.group_av_admin`` | | | ++----------------------+----------------------+----------------------+ +| ` | ``ir.attachment`` | Manage quarantined | +| `spp_attachment_av_s | | files | +| can.group_av_admin`` | | | ++----------------------+----------------------+----------------------+ + +Extension Points +~~~~~~~~~~~~~~~~ + +- Override ``ir.attachment._scan_for_malware()`` to customize scan logic + or add pre/post-scan hooks +- Inherit ``spp.av.scanner.backend`` and extend ``scan_binary()`` to + support additional antivirus engines +- Override ``ir.attachment._quarantine()`` to add custom quarantine + handling or external storage + +Dependencies +~~~~~~~~~~~~ + +``base``, ``mail``, ``queue_job``, ``spp_encryption``, ``spp_security`` + +External: ``pyclamd`` (Python library for ClamAV integration) + +**Table of contents** + +.. contents:: + :local: + +Bug Tracker +=========== + +Bugs are tracked on `GitHub Issues `_. +In case of trouble, please check there if your issue has already been reported. +If you spotted it first, help us to smash it by providing a detailed and welcomed +`feedback `_. + +Do not contact contributors directly about support or help with technical issues. + +Credits +======= + +Authors +------- + +* OpenSPP.org + +Maintainers +----------- + +.. |maintainer-jeremi| image:: https://github.com/jeremi.png?size=40px + :target: https://github.com/jeremi + :alt: jeremi +.. |maintainer-gonzalesedwin1123| image:: https://github.com/gonzalesedwin1123.png?size=40px + :target: https://github.com/gonzalesedwin1123 + :alt: gonzalesedwin1123 +.. |maintainer-reichie020212| image:: https://github.com/reichie020212.png?size=40px + :target: https://github.com/reichie020212 + :alt: reichie020212 +.. |maintainer-emjay0921| image:: https://github.com/emjay0921.png?size=40px + :target: https://github.com/emjay0921 + :alt: emjay0921 + +Current maintainers: + +|maintainer-jeremi| |maintainer-gonzalesedwin1123| |maintainer-reichie020212| |maintainer-emjay0921| + +This module is part of the `OpenSPP/openspp-modules `_ project on GitHub. + +You are welcome to contribute. diff --git a/spp_attachment_av_scan/__init__.py b/spp_attachment_av_scan/__init__.py new file mode 100644 index 00000000..0650744f --- /dev/null +++ b/spp_attachment_av_scan/__init__.py @@ -0,0 +1 @@ +from . import models diff --git a/spp_attachment_av_scan/__manifest__.py b/spp_attachment_av_scan/__manifest__.py new file mode 100644 index 00000000..0699c6fe --- /dev/null +++ b/spp_attachment_av_scan/__manifest__.py @@ -0,0 +1,35 @@ +{ # pylint: disable=pointless-statement + "name": "OpenSPP Attachment Antivirus Scan", + "category": "OpenSPP", + "version": "19.0.1.0.0", + "sequence": 1, + "author": "OpenSPP.org", + "website": "https://github.com/OpenSPP/openspp-modules", + "license": "LGPL-3", + "development_status": "Beta", + "maintainers": ["jeremi", "gonzalesedwin1123", "reichie020212", "emjay0921"], + "depends": [ + "base", + "mail", + "job_worker", + "spp_encryption", + "spp_security", + ], + "external_dependencies": { + "python": ["pyclamd"], + }, + "data": [ + "security/security.xml", + "security/ir.model.access.csv", + "data/av_scanner_data.xml", + "data/quarantine_cron.xml", + "views/av_scanner_backend_views.xml", + "views/ir_attachment_views.xml", + ], + "assets": {}, + "demo": [], + "images": [], + "application": False, + "installable": True, + "auto_install": False, +} diff --git a/spp_attachment_av_scan/data/av_scanner_data.xml b/spp_attachment_av_scan/data/av_scanner_data.xml new file mode 100644 index 00000000..65bb19b5 --- /dev/null +++ b/spp_attachment_av_scan/data/av_scanner_data.xml @@ -0,0 +1,15 @@ + + + + + Default ClamAV Scanner + 10 + clamd_socket + + /var/run/clamav/clamd.sock + localhost + 3310 + 100 + 60 + + diff --git a/spp_attachment_av_scan/data/quarantine_cron.xml b/spp_attachment_av_scan/data/quarantine_cron.xml new file mode 100644 index 00000000..d38199bf --- /dev/null +++ b/spp_attachment_av_scan/data/quarantine_cron.xml @@ -0,0 +1,37 @@ + + + + + Purge Old Quarantined Files + + code + model._cron_purge_old_quarantined_files() + 1 + days + 1 + + + + + Clean Up Forensic Download Attachments + + code + model._cron_cleanup_forensic_downloads() + 1 + hours + 1 + + + + + spp_attachment_av_scan.quarantine_retention_days + 90 + + + + + spp_attachment_av_scan.forensic_download_retention_hours + 24 + + + diff --git a/spp_attachment_av_scan/models/__init__.py b/spp_attachment_av_scan/models/__init__.py new file mode 100644 index 00000000..0a39496b --- /dev/null +++ b/spp_attachment_av_scan/models/__init__.py @@ -0,0 +1,2 @@ +from . import av_scanner_backend +from . import ir_attachment diff --git a/spp_attachment_av_scan/models/av_scanner_backend.py b/spp_attachment_av_scan/models/av_scanner_backend.py new file mode 100644 index 00000000..df20002b --- /dev/null +++ b/spp_attachment_av_scan/models/av_scanner_backend.py @@ -0,0 +1,326 @@ +import logging + +from odoo import _, api, fields, models +from odoo.exceptions import UserError, ValidationError + +_logger = logging.getLogger(__name__) + +try: + import pyclamd +except ImportError: + pyclamd = None + _logger.warning("pyclamd library not found. Antivirus scanning will not work.") + + +class AVScannerBackend(models.Model): + _name = "spp.av.scanner.backend" + _description = "Antivirus Scanner Backend" + _order = "sequence, name" + + name = fields.Char( + required=True, + help="Name of the antivirus scanner backend", + ) + sequence = fields.Integer( + default=10, + help="Sequence for ordering scanner backends", + ) + backend_type = fields.Selection( + [ + ("clamd_socket", "ClamAV Unix Socket"), + ("clamd_network", "ClamAV Network"), + ], + required=True, + default="clamd_socket", + help="Type of antivirus scanner backend", + ) + is_active = fields.Boolean( + default=True, + help="Whether this scanner backend is active and should be used for scanning", + ) + + # ClamAV settings + clamd_socket_path = fields.Char( + string="ClamAV Socket Path", + default="/var/run/clamav/clamd.sock", + help="Path to ClamAV Unix socket", + ) + clamd_host = fields.Char( + string="ClamAV Host", + default="localhost", + help="Hostname or IP address of ClamAV daemon", + ) + clamd_port = fields.Integer( + string="ClamAV Port", + default=3310, + help="Port number of ClamAV daemon", + ) + + # Limits + max_file_size_mb = fields.Integer( + string="Max File Size (MB)", + default=100, + help="Maximum file size to scan in megabytes. Files larger than this will be skipped.", + ) + scan_timeout_seconds = fields.Integer( + string="Scan Timeout (seconds)", + default=60, + help="Maximum time to wait for scan completion", + ) + + # Statistics + last_connection_test_date = fields.Datetime( + string="Last Connection Test", + readonly=True, + ) + last_connection_test_result = fields.Boolean( + string="Last Test Result", + readonly=True, + ) + last_connection_error = fields.Text( + string="Last Connection Error", + readonly=True, + ) + + @api.constrains("max_file_size_mb") + def _check_max_file_size(self): + """Validate max file size is positive.""" + for record in self: + if record.max_file_size_mb <= 0: + raise ValidationError(_("Maximum file size must be greater than zero.")) + + @api.constrains("scan_timeout_seconds") + def _check_scan_timeout(self): + """Validate scan timeout is positive.""" + for record in self: + if record.scan_timeout_seconds <= 0: + raise ValidationError(_("Scan timeout must be greater than zero.")) + + @api.constrains("clamd_port") + def _check_clamd_port(self): + """Validate ClamAV port is in valid range.""" + for record in self: + if record.backend_type == "clamd_network": + if not (1 <= record.clamd_port <= 65535): + raise ValidationError(_("ClamAV port must be between 1 and 65535.")) + + def _get_scanner_instance(self): + """Get pyclamd scanner instance based on backend type. + + Returns: + pyclamd instance or None if unavailable + + Raises: + UserError: If pyclamd library is not installed or connection fails + """ + self.ensure_one() + + if not pyclamd: + raise UserError(_("The pyclamd library is not installed. " "Please install it to use antivirus scanning.")) + + try: + if self.backend_type == "clamd_socket": + _logger.info("Connecting to ClamAV via Unix socket: %s", self.clamd_socket_path) + scanner = pyclamd.ClamdUnixSocket(filename=self.clamd_socket_path) + else: # clamd_network + _logger.info( + "Connecting to ClamAV via network: %s:%s", + self.clamd_host, + self.clamd_port, + ) + scanner = pyclamd.ClamdNetworkSocket( + host=self.clamd_host, + port=self.clamd_port, + timeout=self.scan_timeout_seconds, + ) + + # Test if we can ping the scanner + if not scanner.ping(): + raise UserError( + _( + "Cannot connect to ClamAV daemon. " + "Please check the scanner configuration and ensure ClamAV is running." + ) + ) + + return scanner + + except Exception as error: + _logger.error( + "Failed to connect to ClamAV backend '%s': %s", + self.name, + str(error), + exc_info=True, + ) + raise UserError( + _( + "Failed to connect to ClamAV daemon: %(error)s", + error=str(error), + ) + ) from error + + def scan_binary(self, binary_data, filename=None): + """Scan binary data for malware. + + Args: + binary_data: Binary data to scan + filename: Optional filename for logging purposes + + Returns: + dict with keys: + - status: 'clean', 'infected', 'error', or 'skipped' + - threat_name: Name of detected threat (if infected) + - details: Additional details about the scan result + """ + self.ensure_one() + + if not self.is_active: + _logger.warning("Scanner backend '%s' is not active, skipping scan", self.name) + return { + "status": "skipped", + "threat_name": None, + "details": "Scanner backend is not active", + } + + # Check file size + file_size_mb = len(binary_data) / (1024 * 1024) + if file_size_mb > self.max_file_size_mb: + _logger.info( + "File size (%.2f MB) exceeds maximum (%.2f MB), skipping scan", + file_size_mb, + self.max_file_size_mb, + ) + return { + "status": "skipped", + "threat_name": None, + "details": f"File size ({file_size_mb:.2f} MB) exceeds maximum ({self.max_file_size_mb} MB)", + } + + try: + scanner = self._get_scanner_instance() + + _logger.info( + "Scanning binary data (%.2f MB)%s", + file_size_mb, + f" for file: {filename}" if filename else "", + ) + + # Scan the binary data + scan_result = scanner.scan_stream(binary_data) + + if scan_result is None: + # Clean file + _logger.info("Scan completed: file is clean") + return { + "status": "clean", + "threat_name": None, + "details": "No malware detected", + } + else: + # Infected file + # pyclamd scan_stream returns: {'stream': ('FOUND', 'ThreatName')} + stream_result = scan_result.get("stream", (None, "Unknown")) + threat_name = stream_result[1] if len(stream_result) > 1 else "Unknown" + _logger.warning( + "Malware detected: %s", + threat_name, + ) + return { + "status": "infected", + "threat_name": threat_name, + "details": f"Malware detected: {threat_name}", + } + + except UserError: + # Re-raise UserError (connection issues, etc.) + raise + except Exception as error: + _logger.error( + "Error during malware scan: %s", + str(error), + exc_info=True, + ) + return { + "status": "error", + "threat_name": None, + "details": f"Scan error: {str(error)}", + } + + def test_connection(self): + """Test connection to the scanner backend. + + Returns: + bool: True if connection successful, False otherwise + """ + self.ensure_one() + + try: + scanner = self._get_scanner_instance() + version = scanner.version() + + self.write( + { + "last_connection_test_date": fields.Datetime.now(), + "last_connection_test_result": True, + "last_connection_error": None, + } + ) + + _logger.info( + "Connection test successful for backend '%s'. ClamAV version: %s", + self.name, + version, + ) + + return { + "type": "ir.actions.client", + "tag": "display_notification", + "params": { + "title": _("Connection Successful"), + "message": _( + "Successfully connected to ClamAV daemon.\nVersion: %(version)s", + version=version, + ), + "type": "success", + "sticky": False, + }, + } + + except Exception as error: + error_msg = str(error) + self.write( + { + "last_connection_test_date": fields.Datetime.now(), + "last_connection_test_result": False, + "last_connection_error": error_msg, + } + ) + + _logger.error( + "Connection test failed for backend '%s': %s", + self.name, + error_msg, + ) + + return { + "type": "ir.actions.client", + "tag": "display_notification", + "params": { + "title": _("Connection Failed"), + "message": _( + "Failed to connect to ClamAV daemon:\n%(error)s", + error=error_msg, + ), + "type": "danger", + "sticky": True, + }, + } + + @api.model + def get_active_scanner(self): + """Get the first active scanner backend. + + Returns: + AVScannerBackend record or empty recordset + """ + return self.search([("is_active", "=", True)], limit=1, order="sequence, id") diff --git a/spp_attachment_av_scan/models/ir_attachment.py b/spp_attachment_av_scan/models/ir_attachment.py new file mode 100644 index 00000000..c4ec521e --- /dev/null +++ b/spp_attachment_av_scan/models/ir_attachment.py @@ -0,0 +1,765 @@ +import base64 +import hashlib +import json +import logging + +from odoo import Command, _, api, fields, models +from odoo.exceptions import AccessError, UserError + +_logger = logging.getLogger(__name__) + +QUARANTINE_PROVIDER_PARAM = "spp_attachment_av_scan.quarantine_encryption_provider_id" +QUARANTINE_RETENTION_DAYS_PARAM = "spp_attachment_av_scan.quarantine_retention_days" +DEFAULT_QUARANTINE_RETENTION_DAYS = 90 +FORENSIC_DOWNLOAD_RETENTION_HOURS_PARAM = "spp_attachment_av_scan.forensic_download_retention_hours" +DEFAULT_FORENSIC_DOWNLOAD_RETENTION_HOURS = 24 + + +class IrAttachment(models.Model): + _inherit = "ir.attachment" + + scan_status = fields.Selection( + [ + ("pending", "Pending Scan"), + ("scanning", "Scanning"), + ("clean", "Clean"), + ("infected", "Infected"), + ("error", "Scan Error"), + ("skipped", "Skipped"), + ], + default="pending", + index=True, + help="Status of the malware scan for this attachment", + ) + scan_date = fields.Datetime( + string="Scan Date", + help="Date and time when the file was scanned", + ) + scan_result = fields.Text( + string="Scan Result Details", + help="Detailed scan result in JSON format", + ) + threat_name = fields.Char( + string="Threat Name", + help="Name of the detected threat if file is infected", + ) + is_quarantined = fields.Boolean( + string="Quarantined", + default=False, + index=True, + help="Whether this file has been quarantined due to malware detection", + ) + + # Encrypted quarantine storage fields + quarantine_data = fields.Binary( + string="Encrypted Quarantine Data", + attachment=False, + help="Encrypted copy of the infected file for forensic analysis", + ) + quarantine_hash = fields.Char( + string="File SHA256 Hash", + help="SHA256 hash of the original infected file for verification", + ) + quarantine_date = fields.Datetime( + string="Quarantine Date", + help="Date and time when the file was quarantined", + ) + original_file_size = fields.Integer( + string="Original File Size", + help="Size of the original file in bytes before quarantine", + ) + + # Forensic download tracking + is_forensic_download = fields.Boolean( + string="Forensic Download", + default=False, + index=True, + help="Marks temporary attachments created for forensic analysis downloads", + ) + + @api.model_create_multi + def create(self, vals_list): + """Override create to queue malware scan for binary attachments.""" + attachments = super().create(vals_list) + + # Queue scan for attachments that have binary data + for attachment in attachments: + if attachment.type == "binary" and attachment.datas: + try: + # Queue the scan job + attachment.with_delay( + description=f"Scan attachment {attachment.id} for malware", + priority=20, + )._scan_for_malware() + _logger.info("Queued malware scan for attachment ID %s", attachment.id) + except Exception as error: + _logger.error( + "Failed to queue malware scan for attachment ID %s: %s", + attachment.id, + str(error), + ) + + return attachments + + def write(self, vals): + """Override write to queue scan if binary data is updated.""" + result = super().write(vals) + + if "datas" in vals and not self.env.context.get("skip_av_scan_queue"): + for attachment in self: + if attachment.type == "binary" and attachment.datas: + try: + attachment.with_context(skip_av_scan_queue=True).write( + { + "scan_status": "pending", + "scan_date": None, + "scan_result": None, + "threat_name": None, + "is_quarantined": False, + "quarantine_data": None, + "quarantine_hash": None, + "quarantine_date": None, + "original_file_size": None, + } + ) + attachment.with_delay( + description=f"Scan updated attachment {attachment.id} for malware", + priority=20, + )._scan_for_malware() + _logger.info( + "Queued malware scan for updated attachment ID %s", + attachment.id, + ) + except Exception as error: + _logger.error( + "Failed to queue malware scan for updated attachment ID %s: %s", + attachment.id, + str(error), + ) + + return result + + def _scan_for_malware(self): + """Scan attachment for malware using configured antivirus backend.""" + self.ensure_one() + + if self.type != "binary" or not self.datas: + _logger.info( + "Skipping scan for attachment ID %s (not binary or no data)", + self.id, + ) + self.write( + { + "scan_status": "skipped", + "scan_date": fields.Datetime.now(), + "scan_result": json.dumps( + { + "status": "skipped", + "reason": "Not a binary attachment or no data", + } + ), + } + ) + return + + scanner_backend = self.env["spp.av.scanner.backend"].get_active_scanner() + if not scanner_backend: + _logger.warning("No active antivirus scanner backend configured") + self.write( + { + "scan_status": "skipped", + "scan_date": fields.Datetime.now(), + "scan_result": json.dumps( + { + "status": "skipped", + "reason": "No active scanner backend configured", + } + ), + } + ) + return + + try: + self.write({"scan_status": "scanning"}) + + # Decode binary data once and reuse to avoid race conditions + # In Odoo, self.datas is always base64-encoded (str or bytes) + # Ensure raw_binary_data is always bytes to avoid TypeError in scan_binary() + raw_binary_data = base64.b64decode(self.datas) if self.datas else b"" + + scan_result = scanner_backend.scan_binary( + raw_binary_data, + filename=self.name, + ) + + scan_status = scan_result.get("status", "error") + threat_name = scan_result.get("threat_name") + + vals = { + "scan_status": scan_status, + "scan_date": fields.Datetime.now(), + "scan_result": json.dumps(scan_result), + "threat_name": threat_name, + } + + if scan_status == "infected": + vals["is_quarantined"] = True + self.write(vals) + # Pass binary_data to avoid re-reading from self.datas (race condition fix) + self._quarantine(binary_data=raw_binary_data) + self._notify_security_admins() + else: + self.write(vals) + + _logger.info( + "Malware scan completed for attachment ID %s: %s", + self.id, + scan_status, + ) + + except Exception as error: + _logger.error( + "Error scanning attachment ID %s for malware: %s", + self.id, + str(error), + exc_info=True, + ) + self.write( + { + "scan_status": "error", + "scan_date": fields.Datetime.now(), + "scan_result": json.dumps( + { + "status": "error", + "error": str(error), + } + ), + } + ) + + def _get_quarantine_encryption_provider(self): + """Get or create the encryption provider for quarantine storage.""" + ICP = self.env["ir.config_parameter"].sudo() + provider_id = ICP.get_param(QUARANTINE_PROVIDER_PARAM) + + if provider_id: + provider = self.env["spp.encryption.provider"].sudo().browse(int(provider_id)) + if provider.exists(): + return provider + + provider = self.env["spp.encryption.provider"].sudo().search([("type", "=", "jwcrypto")], limit=1) + + if not provider: + _logger.warning( + "No encryption provider configured for quarantine. " "Files will be quarantined without encryption." + ) + return None + + ICP.set_param(QUARANTINE_PROVIDER_PARAM, str(provider.id)) + return provider + + def _calculate_file_hash(self, binary_data): + """Calculate SHA256 hash of file data.""" + if isinstance(binary_data, str): + binary_data = base64.b64decode(binary_data) + return hashlib.sha256(binary_data).hexdigest() + + def _quarantine(self, binary_data=None): + """Quarantine infected file with encrypted storage. + + Args: + binary_data: Raw binary data to quarantine. If not provided, reads from self.datas. + Passing binary_data avoids race conditions when data may change. + """ + self.ensure_one() + + if self.scan_status != "infected": + return + + _logger.warning( + "Quarantining infected attachment ID %s (threat: %s)", + self.id, + self.threat_name or "Unknown", + ) + + try: + # Use provided binary_data to avoid race condition, or fall back to self.datas + if binary_data is None: + binary_data = base64.b64decode(self.datas) if self.datas else b"" + if not binary_data: + _logger.warning("No binary data to quarantine for attachment ID %s", self.id) + return + + file_hash = self._calculate_file_hash(binary_data) + file_size = len(binary_data) + + encryption_provider = self._get_quarantine_encryption_provider() + + if encryption_provider: + try: + encrypted_data = encryption_provider.encrypt_data(binary_data) + quarantine_data = base64.b64encode(encrypted_data) + _logger.info( + "Encrypted quarantined file for attachment ID %s (hash: %s)", + self.id, + file_hash, + ) + except Exception as enc_error: + _logger.error( + "Failed to encrypt quarantined file for attachment ID %s: %s", + self.id, + str(enc_error), + ) + quarantine_data = None + else: + quarantine_data = None + + self.with_context(skip_av_scan_queue=True).write( + { + "is_quarantined": True, + "quarantine_data": quarantine_data, + "quarantine_hash": file_hash, + "quarantine_date": fields.Datetime.now(), + "original_file_size": file_size, + "datas": False, + } + ) + + _logger.warning( + "Quarantined attachment ID %s - Original data removed, " + "encrypted backup stored (hash: %s, size: %d bytes)", + self.id, + file_hash, + file_size, + ) + + except Exception as error: + _logger.error( + "Error during quarantine of attachment ID %s: %s", + self.id, + str(error), + exc_info=True, + ) + self.with_context(skip_av_scan_queue=True).write( + { + "is_quarantined": True, + "quarantine_date": fields.Datetime.now(), + } + ) + + def _notify_security_admins(self): + """Send notification to security administrators about infected file.""" + self.ensure_one() + + if self.scan_status != "infected": + return + + _logger.warning( + "Infected file detected - Attachment ID: %s, Threat: %s", + self.id, + self.threat_name or "Unknown", + ) + + try: + security_group = self.env.ref("spp_attachment_av_scan.group_av_admin", raise_if_not_found=False) + if not security_group: + _logger.warning("AV Admin group not found, cannot notify administrators") + return + + admin_users = security_group.user_ids + + if not admin_users: + _logger.warning("No users in AV Admin group to notify") + return + + # Send notification to admin users via internal message + message_body = _( + "

Malware Detected in Attachment

" + "
    " + "
  • File: %(name)s
    • " + "
  • Threat: %(threat)s
    • " + "
  • Hash: %(hash)s
    • " + "
  • Date: %(date)s
    • " + "
  • Status: Quarantined and encrypted
    • " + "
", + name=self.name, + threat=self.threat_name or "Unknown", + hash=self.quarantine_hash or "N/A", + date=self.scan_date, + ) + + # Create notification for each admin user + self.env["mail.message"].sudo().create( + { + "message_type": "notification", + "subject": _("Security Alert: Infected File Detected"), + "body": message_body, + "partner_ids": [Command.set(admin_users.mapped("partner_id").ids)], + "model": "ir.attachment", + "res_id": self.id, + } + ) + + _logger.info( + "Notified %d security administrators about infected file", + len(admin_users), + ) + + except Exception as error: + _logger.error( + "Failed to notify security admins about infected file: %s", + str(error), + ) + + def _check_av_admin_access(self): + """Check if current user has AV Admin access.""" + self.ensure_one() + if not self.env.user.has_group("spp_attachment_av_scan.group_av_admin"): + raise AccessError(_("Only Antivirus Administrators can perform this action.")) + + def action_restore_quarantined(self): + """Restore a quarantined file (for false positives). AV Admin only.""" + self.ensure_one() + self._check_av_admin_access() + + if not self.is_quarantined: + raise UserError(_("This attachment is not quarantined.")) + + if not self.quarantine_data: + raise UserError(_("No encrypted backup available for this quarantined file.")) + + encryption_provider = self._get_quarantine_encryption_provider() + if not encryption_provider: + raise UserError(_("No encryption provider configured. Cannot decrypt the file.")) + + try: + encrypted_data = base64.b64decode(self.quarantine_data) + decrypted_data = encryption_provider.decrypt_data(encrypted_data) + + # Validate decrypted file size matches original + if self.original_file_size and len(decrypted_data) != self.original_file_size: + raise UserError( + _( + "File size mismatch. Expected %(expected)d bytes, got %(actual)d bytes.", + expected=self.original_file_size, + actual=len(decrypted_data), + ) + ) + + restored_hash = self._calculate_file_hash(decrypted_data) + if restored_hash != self.quarantine_hash: + raise UserError(_("File integrity check failed. The restored file hash does not match.")) + + # SECURITY: sudo() is intentional - AV admin access verified via _check_av_admin_access() + # nosemgrep: semgrep.odoo-sudo-without-context + self.sudo().with_context(skip_av_scan_queue=True).write( + { + "datas": base64.b64encode(decrypted_data), + "is_quarantined": False, + "scan_status": "clean", + "quarantine_data": False, + "quarantine_hash": False, + "quarantine_date": False, + "original_file_size": False, + "threat_name": False, + "scan_result": json.dumps( + { + "status": "restored", + "reason": "Manually restored by administrator", + "restored_by": self.env.user.name, + "restored_date": fields.Datetime.now().isoformat(), + } + ), + } + ) + + _logger.warning( + "Quarantined attachment ID %s restored by user %s", + self.id, + self.env.user.name, + ) + + return { + "type": "ir.actions.client", + "tag": "display_notification", + "params": { + "title": _("File Restored"), + "message": _("The quarantined file has been restored successfully."), + "type": "success", + "sticky": False, + }, + } + + except UserError: + raise + except Exception as error: + _logger.error( + "Failed to restore quarantined attachment ID %s: %s", + self.id, + str(error), + exc_info=True, + ) + raise UserError(_("Failed to restore file: %(error)s", error=str(error))) from error + + def action_download_quarantined_for_analysis(self): + """Download quarantined file for forensic analysis. AV Admin only.""" + self.ensure_one() + self._check_av_admin_access() + + if not self.is_quarantined: + raise UserError(_("This attachment is not quarantined.")) + + if not self.quarantine_data: + raise UserError(_("No encrypted backup available for this quarantined file.")) + + encryption_provider = self._get_quarantine_encryption_provider() + if not encryption_provider: + raise UserError(_("No encryption provider configured. Cannot decrypt the file.")) + + try: + encrypted_data = base64.b64decode(self.quarantine_data) + decrypted_data = encryption_provider.decrypt_data(encrypted_data) + + # Validate decrypted file size matches original + if self.original_file_size and len(decrypted_data) != self.original_file_size: + raise UserError( + _( + "File size mismatch. Expected %(expected)d bytes, got %(actual)d bytes.", + expected=self.original_file_size, + actual=len(decrypted_data), + ) + ) + + download_attachment = ( + self.env["ir.attachment"] + .sudo() + .with_context(skip_av_scan_queue=True) + .create( + { + "name": f"QUARANTINED_{self.name}", + "datas": base64.b64encode(decrypted_data), + "mimetype": "application/octet-stream", + "type": "binary", + "res_model": "ir.attachment", + "res_id": self.id, + # Mark as forensic download for cleanup + "is_forensic_download": True, + "scan_status": "skipped", + "scan_result": json.dumps( + { + "status": "skipped", + "reason": "Forensic download - original quarantined file", + } + ), + } + ) + ) + + _logger.warning( + "Quarantined attachment ID %s downloaded for analysis by user %s", + self.id, + self.env.user.name, + ) + + return { + "type": "ir.actions.act_url", + "url": f"/web/content/{download_attachment.id}?download=true", + "target": "self", + } + + except UserError: + raise + except Exception as error: + _logger.error( + "Failed to download quarantined attachment ID %s: %s", + self.id, + str(error), + exc_info=True, + ) + raise UserError(_("Failed to download file: %(error)s", error=str(error))) from error + + def action_permanently_delete_quarantined(self): + """Permanently delete quarantined file and its encrypted backup. AV Admin only.""" + self.ensure_one() + self._check_av_admin_access() + + if not self.is_quarantined: + raise UserError(_("This attachment is not quarantined.")) + + _logger.warning( + "Permanently deleting quarantined attachment ID %s (hash: %s) by user %s", + self.id, + self.quarantine_hash or "N/A", + self.env.user.name, + ) + + # SECURITY: sudo() is intentional - AV admin access verified via _check_av_admin_access() + # nosemgrep: semgrep.odoo-sudo-without-context + self.sudo().with_context(skip_av_scan_queue=True).write( + { + "quarantine_data": False, + "datas": False, + } + ) + + # nosemgrep: semgrep.odoo-sudo-without-context + self.sudo().unlink() + + return { + "type": "ir.actions.client", + "tag": "display_notification", + "params": { + "title": _("File Deleted"), + "message": _("The quarantined file has been permanently deleted."), + "type": "warning", + "sticky": False, + }, + } + + def action_rescan(self): + """Manually trigger a rescan of the attachment.""" + for attachment in self: + if attachment.is_quarantined: + raise UserError(_("Cannot rescan a quarantined file. Restore it first if needed.")) + + if attachment.type != "binary" or not attachment.datas: + raise UserError(_("Cannot scan attachment: not a binary file or no data present.")) + + attachment.write( + { + "scan_status": "pending", + "scan_date": None, + "scan_result": None, + "threat_name": None, + } + ) + + try: + attachment.with_delay( + description=f"Manual rescan of attachment {attachment.id}", + priority=10, + )._scan_for_malware() + + _logger.info("Queued manual rescan for attachment ID %s", attachment.id) + + except Exception as error: + _logger.error( + "Failed to queue manual rescan for attachment ID %s: %s", + attachment.id, + str(error), + ) + raise UserError(_("Failed to queue rescan: %(error)s", error=str(error))) from error + + return { + "type": "ir.actions.client", + "tag": "display_notification", + "params": { + "title": _("Rescan Queued"), + "message": _("The attachment(s) have been queued for scanning."), + "type": "success", + "sticky": False, + }, + } + + def read(self, fields=None, load="_classic_read"): + """Override read to block access to quarantined files.""" + result = super().read(fields=fields, load=load) + + if fields and "datas" in fields: + for record in result: + attachment = self.browse(record["id"]) + if attachment.is_quarantined: + _logger.warning( + "Blocked access to quarantined attachment ID %s", + attachment.id, + ) + record["datas"] = False + + return result + + @api.model + def _cron_purge_old_quarantined_files(self): + """Scheduled job to purge quarantined files older than retention period.""" + ICP = self.env["ir.config_parameter"].sudo() + retention_days = int(ICP.get_param(QUARANTINE_RETENTION_DAYS_PARAM, DEFAULT_QUARANTINE_RETENTION_DAYS)) + + if retention_days <= 0: + _logger.info("Quarantine purge disabled (retention_days <= 0)") + return + + cutoff_date = fields.Datetime.subtract(fields.Datetime.now(), days=retention_days) + + old_quarantined = self.search( + [ + ("is_quarantined", "=", True), + ("quarantine_date", "<", cutoff_date), + ("quarantine_date", "!=", False), + ] + ) + + if not old_quarantined: + _logger.info("No quarantined files older than %d days to purge", retention_days) + return + + _logger.info( + "Purging %d quarantined files older than %d days", + len(old_quarantined), + retention_days, + ) + + for attachment in old_quarantined: + _logger.info( + "Purging old quarantined attachment ID %s (hash: %s, quarantine_date: %s)", + attachment.id, + attachment.quarantine_hash or "N/A", + attachment.quarantine_date, + ) + attachment.with_context(skip_av_scan_queue=True).write( + { + "quarantine_data": False, + } + ) + + _logger.info("Purged encrypted data from %d old quarantined files", len(old_quarantined)) + + @api.model + def _cron_cleanup_forensic_downloads(self): + """Scheduled job to clean up temporary forensic download attachments. + + Forensic download attachments are temporary files created for admin analysis. + They should be cleaned up after a short retention period (default: 24 hours). + """ + ICP = self.env["ir.config_parameter"].sudo() + retention_hours = int( + ICP.get_param( + FORENSIC_DOWNLOAD_RETENTION_HOURS_PARAM, + DEFAULT_FORENSIC_DOWNLOAD_RETENTION_HOURS, + ) + ) + + if retention_hours <= 0: + _logger.info("Forensic download cleanup disabled (retention_hours <= 0)") + return + + cutoff_date = fields.Datetime.subtract(fields.Datetime.now(), hours=retention_hours) + + old_forensic_downloads = self.search( + [ + ("is_forensic_download", "=", True), + ("create_date", "<", cutoff_date), + ] + ) + + if not old_forensic_downloads: + _logger.info("No forensic download attachments older than %d hours to clean up", retention_hours) + return + + _logger.info( + "Cleaning up %d forensic download attachments older than %d hours. IDs: %s", + len(old_forensic_downloads), + retention_hours, + old_forensic_downloads.ids, + ) + + # Delete all old forensic downloads in batch + old_forensic_downloads.sudo().with_context(skip_av_scan_queue=True).unlink() diff --git a/spp_attachment_av_scan/pyproject.toml b/spp_attachment_av_scan/pyproject.toml new file mode 100644 index 00000000..4231d0cc --- /dev/null +++ b/spp_attachment_av_scan/pyproject.toml @@ -0,0 +1,3 @@ +[build-system] +requires = ["whool"] +build-backend = "whool.buildapi" diff --git a/spp_attachment_av_scan/readme/DESCRIPTION.md b/spp_attachment_av_scan/readme/DESCRIPTION.md new file mode 100644 index 00000000..f7859b84 --- /dev/null +++ b/spp_attachment_av_scan/readme/DESCRIPTION.md @@ -0,0 +1,65 @@ +Scans uploaded file attachments for malware using ClamAV antivirus engine. Automatically queues scans for binary attachments on create or update using queue_job. Quarantines infected files by encrypting them with spp_encryption and removing original data. Provides forensic tools for security administrators to restore false positives or download quarantined files for analysis. + +### Key Capabilities + +- Auto-scan binary attachments on upload or update via background jobs +- Quarantine infected files with encrypted backup and SHA256 hash verification +- Block read access to quarantined attachment data +- Manual rescan, restore, forensic download, and permanent deletion of quarantined files +- Notify security administrators when malware is detected +- Scheduled cleanup of old quarantined files and forensic downloads +- Support ClamAV via Unix socket or network connection + +### Key Models + +| Model | Description | +| ------------------------ | -------------------------------------------------------- | +| `spp.av.scanner.backend` | Configures ClamAV connection (socket/network) and limits | +| `ir.attachment` | Extended with scan status, threat name, and quarantine | + +### Configuration + +After installing: + +1. Navigate to **Settings > Administration > Antivirus Scanners** +2. Create a scanner backend with ClamAV connection details (default: `/var/run/clamav/clamd.sock`) +3. Click **Test Connection** to verify ClamAV is running +4. Set **Active** to enable scanning +5. Configure system parameters: + - `spp_attachment_av_scan.quarantine_encryption_provider_id`: Encryption provider for quarantine + - `spp_attachment_av_scan.quarantine_retention_days`: Days before purging quarantined files (default: 90) + - `spp_attachment_av_scan.forensic_download_retention_hours`: Hours before cleaning forensic downloads (default: 24) + +### UI Location + +- **Scanner Configuration**: Settings > Administration > Antivirus Scanners +- **Quarantined Files**: Settings > Technical > Security > Quarantined Files +- **Attachment Forms**: Scan status and quarantine actions appear in "Antivirus Scan" section + +### Tabs + +**Scanner Backend form** (`spp.av.scanner.backend`): + +- **Connection Settings**: Unix socket or network configuration for ClamAV +- **Connection Status**: Last connection test results and error details + +### Security + +| Group | Model | Access | +| --------------------------------------- | ------------------------ | --------------------------- | +| `base.group_user` | `spp.av.scanner.backend` | Read | +| `base.group_user` | `ir.attachment` | Read scan status | +| `spp_attachment_av_scan.group_av_admin` | `spp.av.scanner.backend` | Full CRUD | +| `spp_attachment_av_scan.group_av_admin` | `ir.attachment` | Manage quarantined files | + +### Extension Points + +- Override `ir.attachment._scan_for_malware()` to customize scan logic or add pre/post-scan hooks +- Inherit `spp.av.scanner.backend` and extend `scan_binary()` to support additional antivirus engines +- Override `ir.attachment._quarantine()` to add custom quarantine handling or external storage + +### Dependencies + +`base`, `mail`, `queue_job`, `spp_encryption`, `spp_security` + +External: `pyclamd` (Python library for ClamAV integration) diff --git a/spp_attachment_av_scan/security/ir.model.access.csv b/spp_attachment_av_scan/security/ir.model.access.csv new file mode 100644 index 00000000..927e9abe --- /dev/null +++ b/spp_attachment_av_scan/security/ir.model.access.csv @@ -0,0 +1,3 @@ +id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink +access_spp_av_scanner_backend_admin,spp.av.scanner.backend admin,model_spp_av_scanner_backend,group_av_admin,1,1,1,1 +access_spp_av_scanner_backend_user,spp.av.scanner.backend user,model_spp_av_scanner_backend,base.group_user,1,0,0,0 diff --git a/spp_attachment_av_scan/security/security.xml b/spp_attachment_av_scan/security/security.xml new file mode 100644 index 00000000..9a302397 --- /dev/null +++ b/spp_attachment_av_scan/security/security.xml @@ -0,0 +1,44 @@ + + + + + Antivirus Administrator + + 20 + + + + + Antivirus Administrator + + + Users who can manage antivirus scanner backends and view scan results + + + + + + AV Scanner Backend: Admin Access + + [(1, '=', 1)] + + + + + + + + + + Attachment: Users can see scan status + + [(1, '=', 1)] + + + + + + + + + diff --git a/spp_attachment_av_scan/static/description/icon.png b/spp_attachment_av_scan/static/description/icon.png new file mode 100644 index 0000000000000000000000000000000000000000..c7dbdaaf1dace8f0ccf8c2087047ddfcf584af0c GIT binary patch literal 15480 zcmbumbyQqU(=SR05Hz?GTnBdsm*BxQ_yEH|aCf%^cefzHo!|s_cXxLSE;*Cueee5y z-&tp!b=SRr%%17JtE+c)byrpYs^*)rqBI&Z5i$%644SOWM^)(ez~2ud0`yw0U6BR- zLb8+j><9z%zUS}fO(NraVi*{>9t(ACCvAmK{3f>6EFe=`V=#-GwH=fi21ZcC%?@N@ z33ehk216`tgy_y&+UdwGOoiyQxE0tG>?FYE7BU_VU^Nd#brTOu6QC)bh%mCC8$XnR zHP{J6?q+Red4B@!uI#I$jJr&Mb9s0>iD<$ zuR+wn_Wv~g)v~hqXCyn2gCkho-3}~7rwVqob#^cT|HI*Lr++h%Z~%jxz^1|+Y#iLo zY(Qpqpdjo2_UP{z|J6a#%}Lf&m<$tn~GKO;D=HTYw;RdpEvGW4C`Plx`;h%^9lV07{*~I*>D8d~7A^Wd; z|IiAu{+(Sbi+@eZKaGFS%71$NYs&sb_}|p>|6Wz5CjU{BowI}0KTE*WgcWQBwg%fc z{Z$hCzm;Ta!tZ3^WCi{&6^U6n{ZAD^*B-wW$Oa-r=f-RbHUl|ZInfDg*!P87jt$pw{;L! zurM(Pfvw2pY|U-RrEP6IKvrN!!N2tX4+V7f|D%KdPxB1jp8uKX|M5a@AiMvz6QE@L z|EyqJ2X$LpD`5$cjSGmJUKMO(3U&ZHFp!(tnh1RqllIVYQ3J`EJCZv)f*pi3#3YP4 zY;_;(mw~W(F*95)Y)WYoZkRgrLS)eSvJR)Y$S4!fK zScE24BMTw?G63}=yN?Nr!v4s(L#bh+ z0QHoB|LYajx?X9+TnwfJwuDj{M>z;4bu|DB7H;cherVEncj0{^h73csRh5-&U)E;4 zNLVpq{=h+rsFoNmYz*8AfN`m{D6C^2%WV~zRAFNZuAXKcKMErci*PnF0ZSfM)erUu zjcjUMJ_wuF3RSJ9O~@Z4hhap;#(_0ma`J>1A0~<{s?m|hcz{e!L&u6Tp}I}Ep<>4f zOJS|^MQ_DPOkz?*AhrH}k<9ZOEt4`FAyRDqXjTP|E_#oO27Gr&f`y5OM@B1VqH_ES zCTweSMCx}a*0xU}@o6fA8_gjjy z2Q57xXmg+m(g6q!aM8mCkithJ--tyXkCjku;FTF{?B>(>FABGzSGUggUumv`+C6Ow zvd1XmI~#j#dG0vl>e;QtxGX?gJsdQ+{-4BuDt%|kxthFj<_dORK@Rc;K*$U=E~?kF zJ$(-vwj?T<5%x2c(fneoKTjS|rpBh!8`&y_y)z)7Hj@j%)+~SkVR8K<@`g&WZjo&G z8?wNoqyeOzOEhl;E4C^_e6^7aF#Fx~(z-&NxzGQQC}?L?Gl>qxwKg;MZTpfMvw^V{ zmT;>h9A?JFxNyIC1IPqQldk82>?{LtnMt2Xo$HmXr3gvbffJCJF_|;ZU)lTX#2_{h zNT=4@taez10pm@hvzTLIAAD(`*Y6XZr7!w3a5sy>KWlOvJ92!fyI0Yjt7_+Syy+$Q z9i0@K!{?>N+F!J-sDJMIV zySlF4rF1c1>K1)CaHBkwkwVV z_lfaZhdgZH%&PK>eJxwrWn!sr5&Gc_9Cr|XDCGA_XN{>#)>Qgl3%Uyi`^M@mPTT`? zf;&`{13;P8O-+u@Hlr4IZO)ivM_w*HE{G3gydPIhU7gTd{}##Tw;S&&d-&?A1qaWy zLlnn3TyAMVFPcpfZ`1wMt^$+g?Z(_ki{MSWsfo#KTB33CzU=9qQnoXtdS(mcmLjCY zalOGBnh*x}*Hy&3cD8}2EUr+55qEqP9$UCvz=o=kb9%C^{(Ki9<6A_yTJAVGBAyn3 zIGGLv4!o55o*J5V_xfbsyPk=kC$C`%S6?3qh!N5V(<2M#9p=&i>al1cGc#6pd37`_ z3RMpN=*|e9{nd~zZKGX@%J-K$=_&@x#D$&<8NApJ?i3jM!5X8abIiAPla~}@BE@Ep zytt_iw|xY%OQxngqE(gy8xY@vUMZuc7&hw5I)$M+5$X^P z;i3S7-Tgw2w#pV1R->>O;O~UyyX#p3>DD8rfL3FNO@kS@Uw?F5(eln`lA5WMkAVwk z6(1gr5%VDf8>tN;vdaPZYs8yBSJ^oba~WDr`qr8Oh#ok4VLQ3lrJrZ_Xm(T@FM0qa z&kxcByGv0F-Fx%t@9vZ7JP$}yAKpn-r^LhBTLwsS1J)bs6T{~SIQ6H$7qanXOrs1*Z5c~M%>RPFWj8X;g2@Lhm?HnEOmg0If6exM<_Fa9>!5P zv6(xpC9c)Yz1{ue6}vOIV(QK_dbu(^ad>yOhx?(?cWg0n`J-318#Q=eVZOiuW}A1? z=YKkEE?wkr+3_PaFv)gRxm)xjwl4{Gcz$5;$RixdVH2Ds+=H?$xTUn`QZ<#!D zWRP4okEG?OLnjctlnTlg5)kz*Yn=}m<^joJPN)}L??y(J86Fk_PaZ`{q?IKql37h; zDKAk4_|={_s%_q*rZ}MznUn?=QC9T$A!MnV>~b~n=uXQdTx6` z)C4lw2Vd8?lJqhAV%eA%mg9eTcNjsG(q@@$etAi9{uE1m1hj1!jelwHV;%czJVoYcrZ=vANJHDiH$G) zek&XC9nl=^c*OxElr7lsK6+aN5c^^)p0n;58u$EC`TpvB9KEV=zK9QdPpmKCHANCK zliMaTnv1|oI8A%NctUtQg)_&D9wYY|Iwm&nkURyL3PVzKxQI{K6C{+zFGk`XQGDw} zv$z(!mCfUPd6h*?RowKmNy|p2Mri1laA2VU*^f5fL8Ne4IPc)ybITH=)f$-My53); zfsHD{N>w!&UkTyOxD>>Ey0g^%;L)A?P_Nyhcd+dwhH5DN?-^*`{IEk;(NK z+#s-OPFRbbX|Uo9=Y@)pgD@SCE!UCmYYVmF+$i4Kgz2lR3|L_DxX-u)DSS39jaf=r zT6deEL2ULQJHvU~(|2vtWZ zLueKkQ*#|Bj9fi4c9{)Y&z^&}>=~e5Y-HCkQ7Mw zXCH5+<@YAqb|zki@0M(%ccdpqTJ62ZPg~bZ9%dCF9k!S%_lroxG?x3NpXG4ZBn}!6 z+=_Y!1xqxCN~6zvXAyVg)}YKk4ib#`<>h_p{S$I>vi*LYB5ST+3mf_t)@{}Ih`};0 z29&^wWHWl>8kd64(wY}#hrVQAh&s7gbeHd|IZAStUZ&PSb3$B{PvD=+ zQkSe%LJ0K>h&Kj#S8^)h9GXvu0IZ=3Z>3DSi8{T;a z0b*muMkNGwF;o1RwtCZDg#97P8vE(~`hga&m%k(gTR6qI^gs7yTIO@ay}Te)Hx6Eg zd%2g}G&u)zqqNrD5nG*q8XFK&z9RjIS(Q6DYG^p!6>M30Ef+5|le|Ud>m9T2((_H@ zmT!+5i$HN{<G+1EEoc4AS9vm>QDZpO>K6M{G^b)txOnqNOvTfV zwR^y>(e?%b$$pu79ydu6M>3?3(>(2u(=dN7HK{92%u6nm^iDzS@)?5XBIF{B#CklVg~i#wA$0R9A~jYSgt2E^Wysxcp!2- zJy+&-mzNYaZTSq9cjqTE4)av2f-f$0H4?(;)nFcK>Cqg8V1?|=v!Y(*^*0|9I;_Rhhiwc^cQM&I zs2P#p?_{f-yhS#$Z%c?knJ_g7Zhv%L*{tf?J?E8j94bImWV|QMY5x(sTCL_62EdT)xWZ#KY;8qi zzh&-cv3YOkp`;b}=k-{kwTe#GjC6kh`OVE6++^#^n`2$=$t@u!WTiOfEEDax{k6!e z@X;4kniF^87>l=U_UXRvHKDfp>vDPBi03g%yHSkk525SM)oqOWGqYp4$RD*p_K`zZ zX5;Tx^`n&DE+;ujb3D5nIv6Mom3jfVZ5mIfq!jf|AhPk0p*BCT0x8R9-BE8{1h;FQswTy?v#0}-38B!kczy{x;$7!io^DZ=IcJY##vEYDk$eMl;r^~T9QM) zQtubaNKNtRwxEV=;ce#Z4d5>nKyB3}bT9N~-_eBgFflJtua+a>1#3WkFbOfK>wALd zZQJFC>tFY+A8cE=I=Kr&9)?klwAYSC8EBln7`QBc`8b2H&Uw!rU@nG`1p+M z_PaAlj^s@QS_#v-S7a>mvT=DTFWy=ZjjGOXi5cF@lwE;85aI6_m*ok~r?Q!5Pm%ZT?$+H*@!&OVYR1ei_3V-7Rug|y! z6$Mw3zfY~M&=eRqCgXBTaB?UI^f`~CMbB=}$Mp5L0V>1!a|Lt#a+4g!0f$6;UDKhZ zlL^j^u4Vmh%}jY4)Cwro5tJ1AQGq1f_B}RfX)D2nMS91)Y;HB$dH?2hjtC#Za)<9l z3Xk+rZ6knNtjm9pc2D}(wY6@|ZX5l(cbwO2oUZoqp~U011TV#IhMJfGfJ%N_y5pEr z$$IA>?#}aHx9?aiZ|z18x!q7sz$jnVblQi|AhW85+>7y6btIi|OvFBI?tT(4eXVCg zeP8}0!iu@r=PR>rJ3wq*!=CC<_ihZL5#EG)I$$%%kh7e$zQ1S@xv6Or7!_P&%MPMk zACVS&BE)NLV(qN8MOV5C`xbf8IbN#MmeEcdWYA$OwFX;!1z7PC6DoHe>+fVejhMzC z1S8qnm<(G9MXIvx3DE3&Qo+7^LNi#xb$$M2LL^jXh)cbb3h%G(i91(WK}lj~^MOAm zA?4cXvn!=%bKJ^P|1)ix8c1H28Z^2L({~B=9);^+7Yn7*L|+tIAJG4NPUMk$gC5&z zQeEbR@FbxHdE`+3^XSBSPAWGx5R7Z8yZbLJA~9Q9x(L@tqt{q61Em+ikqTux8^kZ8DQrK4FB3r5Qx$xHG!>D| zA6?vk{*>E?Mj18vgMk%hzN`ZwTFY1ltHNF5S%);i;&*l-ACcsI3pnD=iX?}s!s}HC z1As^77XFUGAm4O;CtDdaLT6%hOQ>4n&pujtYU7jL7onxKBM-_>lW}>$dS5% z{BRX)SUzjTUq2m{I3;m4ULG3n!EI@PR04_rJlShCF+6IG-&{VfY0G+|OLpY);~Tcs ze2Y)Mw|IXXzocJ3+sL=yh{1EwAusXV3dh~TOl+|FVY|@xU{j6Ef?(e4;reCW_43yL z<76IskRMUIl)Uop?JzOW;#+p#(crQzC^Ot~KFDqBhT`=!Rk%4%b1(y9h4j`weN&J! zbyYm>{7aU7#kdNy2Zqx-hUyr=|4NbL%;CXS<-w%jL)X z(3_2Lz*r;mD9!Y`&iV2=x+?sNv)b*Cwn}{YDuYzmi4vn!c+r}V?AzoFZAreI-4!3+ zY{Td}nm@04BAKyM->B1)oKRD#r|^W|jYVjcSAs1YI=xx>$jpFe*KbLKby=*pW)eFs z3ZSXO09)sD}&}V6ipbE(Y~?r$YTn{V-9};R(?Z6wH9Dqxnt8t&~=!h3e%FyMY4}MkN68X-2kX^|Im5y$c6sN{v&x4l_54O-p{PrDCP` zpOp-`$#WIx;mb_%^9f@!#b^Gv=)X8dl(G-ESKr#_UVal#eY9!`MLqLs4DUCH##vQR z*2n?o*KjGB*u!M&?xGOuHa@Hn5s811Ma6+Zz~-qI^cWAxkz$M9EYF+65Y<;MSmJ$H zrmYW$Ykr63;#?@3U~a9Yw$VB(W+T|LSC!M@RS~PJ#aBNlsh@MN)U_GZ+y4ALdVH-Z zeZ7rMl*xi!f6B*qX6Hr-YTWI3@7e|R;u4nUs>YIecpOF-fke*=0lHfETe!@N?>>DK zH=;xe|L}n!7YQPC**{jgAE6=E{~Z{`{~?;C(Z&12K1p^KRB#YWTRU?2RV!>AocDk%*gKH;(HiW`{1C zLgUncZHb`P0zyddG&COjHi2(%mgVv|gu%=`hPvnQickVe$8=lkQe4}&0*&it^=Vd~ zVz5rO$n;=raC-!!5NB|-XZOI{gu$ai!cKY`c7x4qn^>9w9*^aS`tLIdSOvMcwHy)z zisz9h?)wgaHN^ZNO1m|OBga`a*37=gS%}sQp9b3`#|ZInRQKnNUU+Pz_?9%$FWdS@ zDK<8SL9C$=vFNfCZZ*J(vU|VM+)OqeUmu(7t6G4CEYvRUzK*`Qc@f3dneu^f+iG!g zxv+3dL+uJwWvD@yd7%RLmAuTRViISB>GdFBTIdcF28A`w;mJ|!FUG!hkwvww>N>lf z{H={Dx0PPqaV^{;baO8&Z#4W&_23HA>#O7j4>~jvphax5{G4W932b+Oq40dauN4&f zHNyo<4ks5vV~{U|A^h&ku)Ss;0}g#CCAB3 zx!5?ck zw{=3Qkp*j2pk4kf)hQYui~#aNqul$soANTlEt(Bg?n5v;dVgpctq zgK8zA*my$SKTIf^aU6WAcAVx*VfEg7ZkR4Xkr@Rqgp~nl)WKhG;{9Wdad0u6&{I#2 zxKYvs;M&vr=pb8WY#((GbJMo#x zxUcc)yW;DGO<4}gi6di1&45IQZgY_)!A;*)F;lrKSVH5fXFw*)gR$$6cTNB0*>AV^ zw*?Qj?T1Fkol|$DCNdN;)9*Q?6o(#96gu%a7X>rtoCf7n-ECFW5M|6Fal%oQ_HyFT88UEWBj-cYRmoJO?h1i zO8Pb`owZMsyI;28tb{Eo<>GSuU*PNNxjvSV(T~f_NvO^Dd~+Bv4RFyUso1bz_tFj% zCD1oMN-R7Ol)jcmv3xpONAc4_)~6O6({Dh!!AVxU&q++=$T73FoVhi&?s_pYN1!5s zSLaZGTy$Mp1n=}=+x6NJ7#4%I%HoA<%SY4XdQFZO;2iFiQP0678T*1q9`dllr^)b=7CHG-dsj-%14Er*pm zRd^>8M#r;=H+aYIt_QD=wbxFhWWMQQ>)ENMK;y%e z-Iu6Jt^6|6l4x)u>Ylp;h!pn4O+sEjgtk(?U5Hp84IOs(ACPd#;dKgps1N!cG}yQ-Gvsh`Zg?5UQf#j}u^uV0^fBdXFH8Osx2Rn>nD?ts=VM5s(?3r8fR! zJ`WX_!j}fLK<(%2=>n7ezAMSisdM;Al^QJ_vPLj;mPAD$I~PIuyU==s!xUY zodiCv+RDXwU$axLZtbz}8BHq_1XqHo-^Kx4+f%NMl&->(9MD7SO zj&Z#}?1hK1F$*vE4Hl-52+kbud@c@%{KDPxs}pYe1D656Fec#qx9+xdyZ42hGFio=?^)UY_>^ z(>JtY69@hM-~dl%4gVj2NS%f*G|0Te8IlHlUZ{1k{U#Aat)_Xldr;o1s3ZVmargPD z;rI1QJ?8u0>5}@tQ>^!bMR8(pgdU-=nVzFZN}3-}d2iu(c}?B!g+r&S-sFg(f%#=% zzo*;ppCC$j0$qWo20Ac8Gv%A07eM$IXBHv$ov2<=J=H-@-^-4pGZ02IribPegl|FT^(ObV6vO4);?$6A_cuA+Vq1WmKIXgG`?%u zrna{Hm7|qSZ2EYj-pae%klBl5e4Y(Q1~p_8K*?L8**B54K6R1iQ(L|wGo#bCl5%MZ z{MaKF{!lpQcY)8@^9p+-R{^~zI?PY8%s*F`Jk24WY@RNKU0ezwO!ekJFkp|~0(i49 z_o5;d+*Sc(Jxsf-=YV#pfx^q|3d>HKjaXhv8upfShP@MxO3ECHoT?wPg+rAJ6j6d% zuauS&I`}i%EghL!ET5Xxwzd97;lDf-pr|@|G8SGFIUE-hbZa?YaLw!-y(k#t(PILzr}1;;g9@KM&6c28i1cn_xi z(F2R>(iI%Xx#oN~+xepmM0U{~Zb-ADBKO>klUgz|STaYC2~5Jw-3Rp*0M~QeAK_ zLT0jdy1u+74qNvm@lVU?i`<{VyiM-Y&YKwl`Xjzk0A)rN&XTzJ%RhJ_zfDfUp6RejT}_&K~L%hzXRUt_YZ--idup z{Yr*e6)6k#)Uosm3Dq!P+F%<1B=Fb-hzMKL%lx|uDvf&tWb2JnpRL}zSR>)WD&oy}+RNe&Hx|`=VR=Wi6 z7&fK)_A2^4+$>xJ4og%N88LV2S%ppZIE zH}jy~y(@yAt|h*1Nxup80`#-q*0us&eb+uNNliaG@F!bj(_qP@^T>u)(1yV%FpQ$n zoKE3aW`7m0ClO~zsXnJn<$2eljws67*~7k}IRJrorv^i1N>PKfyeLy1>m9%`U>1ap zV;J{k2lR8fH=dT%$B_tRpR2BUFNTgQel2SkW5@I})FPn?lSPtXkB>FA*)4J8-*uAW zCj}gqkZb2+L@sJuIUggVf$OL;Y>9EQh7-fNqMs=W2B_3h8cl_69%LDsEY$=;9~~S` zMh@TOiRbWVES8&JU#7~Z$xYEa`to)$0DF2z2*5Lsl*Ex<_be}5`*h@>p^QK!M@P+% z#{3!j79}}Lm5Fr$lPZBYi+=zlA@aChAd_LxVid4#ykJ+4hoZ1$en6D#@EK`u4o>V& zud!SQXGsUrKUS+``^EDi4qnc;`NSp8QTiL1dq1V|9XIXS zV;zJb0ww|#p08c?^r4SaJIza(jxgVH0p`+7SR4;gt3y0wS{a(dC@t93kb(EUJh7r& z7MBx@f$B+}QZfvbYQHp(Lu{6-@=K)G)# z;RhYWAL`WxFppsry{Tk|`?4(3?>~%ESH%KE zvcS^HtZR~v}xc}=m zvR>5rLTBTsUDrd2`cEyI1D3J_?_lI|P-a1-O+Q07RS0!rKToiU|Hn8yPY>0P*kiZc z6(Xfc;fiU?ES|Vm+ks*Vpm_tejb_d-eAbc^lTRL@sJAyiWcR9{&$P+wgPs~tFZ!}l z^6r|Pg5#quRe6tZSsl$ggp}?@@q&MP50oksD}Nwf6Z)+xqSVfwk?b#H5FhXn;mW?g zee;BWj^!4}gGSGiNNN?)^t(tIj;X|PR|DOk=*!w+gnJufT-E(`1wkOySh?PpR^$pf z=C&Fm7Jc|imd4*ZU&i=Zg0L;lkL9lVe!*P|<`G|EeP!OfoDbn!NH&?6Z=CV3jYg|# z?BpJ9lL>ALqBI(XWi4d6aqMAxVmN!5cj;efWj->$d#)NEJJ#<|R^9vcL-0&M-$#eJ zzrJyDNSoZz;=rD3V-miQ`OdMVdl2YHgHr|zD}9~CE)C84Tc1J1$`$3U&wl93G=jXD zZ9mA>7Sd(Tk3uUEial1UOn+{wlLde%u+wNNp8GgWG9I7a!G8;4$o z&2Ar8?dKiphR(Scds1)b80|OkURQWunL*dL1lfeu=EcspYtvf6+Di-L{;zd;19Afh z3TKDBiw*7_i^M3@x(AL@A~gpKShwgYD^G=;gxS8@9O=!cILWlyvqzha!M_d-1^uHa z0?SWjk&$Rw%}0NVm|eELTYj+3)|1iojv8};RmX+q5PG0x0z#`}9+*fyQ2{%ps7U;nnT3i34#>rSn2@(?>~%+MK$^b;eyk>j`K;Pxxt zUp)+`Wwxnw)l0~pdDmBNFbxO1%N1e|?`#a-wevf4WLUA6I)pOIM44FJ_75}Y7% za<*RY2Q7gH&(-O~t*m~}u&qGlDp4yW*3(ZHUi^}OdM%SXXPZjGZG(Utpil0LdTTRnCpSa}-t+SE`GR5a05{VN*n65{~ zi+7QCL&nSPW{W|;T=bXC(S}yeza@Zb%Y}M>bqdbK(|tE@kxUAbk*YcsUAYWuYwGL8 zXSK~8GsGO2jDT6{A~I|(i?tJVY;~Ikn%nJ5=u=PiI!-cViCVec8O4!_tVPC3-)Ziu z0Zoc+qud@e>ES`yL()+w8?FNF%<&fKS}whZL<|P!ZzL-mEZ?rOr|+*v^0EA!)!E~O_ba%&;*9IA zolizsa!TimzSm(GUWK++qz=+Ik&+@820c#?Ztm%XCE>V2FG1_;7W{V>WIW-d<~qN> z{)|8qXh!q-b2TG1AMYIt@65s?DEzUAV}}1r(M|F5F1#~WsH5)G2VY3OLi&0;my9QM zL);fdhGxx5^-4^Cd$-&mgc9N1BdV&j%1ih|7-dd@-0mFO&5E0iP^T<1nt~)(*5+P`KrfMS6pkxSQoNXO}tH@;S*V@zdXcUsE&Qh zkoX)6{0fsMPULHE!|ZD>_SPqK?8M}^w1UeW_$&2kT$zqS{*Dl$>2{rq^AAKf+$3I4 zslbVh%{kmT=4(zI?%M8hIVBDV0c+GUi)Gr*qmoMBmxR}%K_R8vtBq0#&Ln<8D%dwN zX>kpAbVWC%Ox9N${Hjz6(^5A2n+f1Ik0GeHcLj`&aX>$e34*En8Q{+qdkxN`e0P!Q zuT;iYl}dM4*Q0MgBHJ<84@Drs)lj-ad^2LCL9)}-LW5l0bPW}DSE?e=%7tHRP6c!f zCP99CfJmiG!~WA`Zs>WX_>h?A{&2eO`K0L$B~4a>l4;-RvWE$eh*xW9ls}c*r%2m& zhNbWPIhO^{^mI=usAMI#22L*o5en8{Hbu|a4~HQ9hIR0+{~&iYEP}?yfr8%s`I47J zMwZl{wRZeoXI={s$a<8gt4*Hsx&iJrQu%P^vb{~RDum$htr@A?>pqxhJV!|gGX zUL`*6%=J@W#QW;LfrYA4&d56JDBXjn3uVUsl49ZLp=uN_rZPtr;;F^iL`u&7(bYYE z=-J{N7h1bT#haD>N0mi%ys9&r^nC9XKh(_H-B%M1HioTc@Fodl-(@UPAvoeevvF5M z;u?_+EkqcJFxApR&f{>;#tk41X4PLBpc|{$-TFD}ZVekXDPVQ+63XB7XBQ-8=C;P3 z^%)ycbSmcLP%&N(tleOR42l01d>VaW(oyOFt;?XYt}bL$;8)^3M}APjS8m#_k+KnP z&zhAc!sRm}|8kYN?tC#ptdd*2*cMd_z!=a0ogK@^%YBXyrw*k^hJhtb)UY-Pp|U`b z;vm3-f2h$+A&q7+M}Mg-r9>2BEm^YPNmZ( z*7I4&!nFAzxpw5$n0?QdSE`^*s^a6@SRrre`i+>=SLtxw^z-@jraYqw@bSip;u!dK zTL9hZVjx|G5={P{9@`(L2W{{d>D%clZO4f70pf2!tc#MFU?)YLt-?Z9$-c2McL4VN z7W9D4WMOAN+6=I1Dfa)8xF9t6=O(>9LB!e%vOnrk?M0> zhwcO)UQOE|!|+=@H*wsyK!gv02uY?=#%_C5C4PYHuGzw%hucEDs@DbbO_Caz!aR{U z+)TI!k?P4(-i%WA5m2zQmZE^K6<+p?B|X5EGq$zw9(PfkANGFIjOw2MWg zKzz_(5iAbl)Py69NJEsQh^vxIDgheWS-`flG+rfqdEJahS*YUq)RCw7wJ6IA7i?_T zbD!-Qf(p&XhfA-KFoYvL#L~7U6T{tD%|dbL)o=N6;2}mx z!H~)1Fa$U)<8*lRd8*EEO<$_82C=Yv=lCg~$8GQ49)$Nx5fJEEaxF zl)u`I99^+<`OtY7_q=-`^1k9=uN<@9D*Adv0Q2^am|DSo=F?vA0J6!bIBOyEjpJ@H z2*UlQP-z!NN@6biXcIsE-B$>G4p#Bsxw4!W^oDs9n-adqf1greR# zfARMgj5m9@`A}9Oc~h#WMos)V%?-=nk`+S6=Q3Tqj&FJVY_lXU-j8{UUQwRer*vNi zfYU!5rO0Ef|MdN1vc@5-WmGYcr9CI@`kiQ7RL+ztb22U{WAeB5;o6w`-4GP9`W`>S z&_}b=Tjc-o#5;+YZe(ff8d~EuaP3uP~tc86jg5qVOZ*{cwJGU z(V#giqR;*#}M7(H=WegGj8QE45StkwQ)t zkDqA#;#%akszszb-d6hC&Y>(@IusF!_+GjwxIeDH(7}w)oA7)sg+;iwNG45>Jl=*4 znht+k)I22GMQiXwNWP<7d0VRrHC&g~daE&5*a?1)=?cFU!1v)>Lhov{i~V|%SV+9X z7((>eXMfQ-lj3T*{T)ezIo7*te0-jq5m672Z%@7nd89JjVZ=_Bbo1hLe3vR5GZ8VQK$3BS3rpv(TI z*if``DGY`pJFPa|qyC_%M6lc!v`aS!?Bf{jCRy3h2>YLHBX-_Z0cNP-YKG+9aVn&&bOWM*j$kk8_d6 z?(xLgln?2|OMK3fRpgLJC=#$Sl$ZdT<~F@JI^%N{SsMK=7C#~w8JCp|ODKUjfulX0 zRNnimv2(P`!_|JMWw#2*v%0*WmV!FHXJnXm$FI8bV27U>i%M0TS`CxQy!TVI4+Hku zCU|>U(96OE+nSptiO19IE`KjZoFmE96%r=Y#&G77AMX8@(Ad$co7FH1**~KH7%QV@ zq2D^0XG`W%Kwy))B%jtc34_bP*&~!hvXkx2x61x?cm8VL+eR&j+qieTj zPcf!P__24db-NUOd7qw4jxNS~Rn}k`w;L-!+JMkh*E38;hxBHxU%E}SZ(^oQnTt9( z6U*##{JUsmtt^A>6&UNN5mxBooYco1=6i8#6YtoyZl1O{hP>>^Lrts-xuXYNTZ$>u zpfaVW+VhuTa-W6(Y5#`hX)X;5E-i}{XxWY&i-0|tDN1{4YkvF|i+8ibuT!lOje;w< zkwW?d17jC~Qo*}a1btjLC$U87&ALRfBUk{XiT&dcIexY(=W<~(r-<*5(6%;&Rm^bw z25DIcIe0Kk;h0MuZVN`^O#>~4>J*7fwa5457~M`DW}CLMhrohubV?aHB0*q%i?F@) zYwum|^K0)Lu4E}LYfhYog~=@Pv>I86X>U>2n?#DFw@m4G^1i2s(0@%DkwFgxASub&ET6!HG@u+jB+p_yO(GoOV3#Nw9K0GZvg&5PWug{2eB{b>*22oK9 zncm+N91?M1gpr#Brp6}vt7WNs#8Bn}aw1X4oh$4)t6v( zbHB1*nkJIRlGpzHfGgQqz$g + + + + +OpenSPP Attachment Antivirus Scan + + + +
+

OpenSPP Attachment Antivirus Scan

+ + +

Beta License: LGPL-3 OpenSPP/openspp-modules

+

Scans uploaded file attachments for malware using ClamAV antivirus +engine. Automatically queues scans for binary attachments on create or +update using queue_job. Quarantines infected files by encrypting them +with spp_encryption and removing original data. Provides forensic tools +for security administrators to restore false positives or download +quarantined files for analysis.

+
+

Key Capabilities

+
    +
  • Auto-scan binary attachments on upload or update via background jobs
    • +
  • Quarantine infected files with encrypted backup and SHA256 hash +verification
    • +
  • Block read access to quarantined attachment data
    • +
  • Manual rescan, restore, forensic download, and permanent deletion of +quarantined files
    • +
  • Notify security administrators when malware is detected
    • +
  • Scheduled cleanup of old quarantined files and forensic downloads
    • +
  • Support ClamAV via Unix socket or network connection
    • +
+
+
+

Key Models

+ ++++ + + + + + + + + + + + + + +
ModelDescription
spp.av.scanner.backendConfigures ClamAV connection +(socket/network) and limits
ir.attachmentExtended with scan status, threat +name, and quarantine
+
+
+

Configuration

+

After installing:

+
    +
  1. Navigate to Settings > Administration > Antivirus Scanners
    2. +
  2. Create a scanner backend with ClamAV connection details (default: +/var/run/clamav/clamd.sock)
    3. +
  3. Click Test Connection to verify ClamAV is running
    4. +
  4. Set Active to enable scanning
    5. +
  5. Configure system parameters:
      +
    • spp_attachment_av_scan.quarantine_encryption_provider_id: +Encryption provider for quarantine
      • +
    • spp_attachment_av_scan.quarantine_retention_days: Days before +purging quarantined files (default: 90)
      • +
    • spp_attachment_av_scan.forensic_download_retention_hours: Hours +before cleaning forensic downloads (default: 24)
      • +
    +
    6. +
+
+
+

UI Location

+
    +
  • Scanner Configuration: Settings > Administration > Antivirus +Scanners
    • +
  • Quarantined Files: Settings > Technical > Security > Quarantined +Files
    • +
  • Attachment Forms: Scan status and quarantine actions appear in +“Antivirus Scan” section
    • +
+
+
+

Tabs

+

Scanner Backend form (spp.av.scanner.backend):

+
    +
  • Connection Settings: Unix socket or network configuration for +ClamAV
    • +
  • Connection Status: Last connection test results and error details
    • +
+
+
+

Security

+ +++++ + + + + + + + + + + + + + + + + + + + + + + + + +
GroupModelAccess
base.group_userspp. +av.scanner.backendRead
base.group_userir.attachmentRead scan status
` +spp_attachment_av_s +can.group_av_admin`spp. +av.scanner.backendFull CRUD
` +spp_attachment_av_s +can.group_av_admin`ir.attachmentManage quarantined +files
+
+
+

Extension Points

+
    +
  • Override ir.attachment._scan_for_malware() to customize scan logic +or add pre/post-scan hooks
    • +
  • Inherit spp.av.scanner.backend and extend scan_binary() to +support additional antivirus engines
    • +
  • Override ir.attachment._quarantine() to add custom quarantine +handling or external storage
    • +
+
+
+

Dependencies

+

base, mail, queue_job, spp_encryption, spp_security

+

External: pyclamd (Python library for ClamAV integration)

+

Table of contents

+ +
+

Bug Tracker

+

Bugs are tracked on GitHub Issues. +In case of trouble, please check there if your issue has already been reported. +If you spotted it first, help us to smash it by providing a detailed and welcomed +feedback.

+

Do not contact contributors directly about support or help with technical issues.

+
+
+

Credits

+
+

Authors

+
    +
  • OpenSPP.org
    • +
+
+
+

Maintainers

+

Current maintainers:

+

jeremi gonzalesedwin1123 reichie020212 emjay0921

+

This module is part of the OpenSPP/openspp-modules project on GitHub.

+

You are welcome to contribute.

+
+
+
+
+ + diff --git a/spp_attachment_av_scan/tests/__init__.py b/spp_attachment_av_scan/tests/__init__.py new file mode 100644 index 00000000..9ccd65a6 --- /dev/null +++ b/spp_attachment_av_scan/tests/__init__.py @@ -0,0 +1,2 @@ +from . import test_av_scanner_backend +from . import test_ir_attachment diff --git a/spp_attachment_av_scan/tests/test_av_scanner_backend.py b/spp_attachment_av_scan/tests/test_av_scanner_backend.py new file mode 100644 index 00000000..c22e908f --- /dev/null +++ b/spp_attachment_av_scan/tests/test_av_scanner_backend.py @@ -0,0 +1,281 @@ +import logging +from unittest.mock import MagicMock, patch + +from odoo.exceptions import ValidationError +from odoo.tests import TransactionCase, tagged + +_logger = logging.getLogger(__name__) + + +@tagged("post_install", "-at_install") +class TestAVScannerBackend(TransactionCase): + """Test cases for AV Scanner Backend model.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.scanner_backend = cls.env["spp.av.scanner.backend"] + + def test_create_scanner_backend(self): + """Test creating a scanner backend.""" + backend = self.scanner_backend.create( + { + "name": "Test Scanner", + "backend_type": "clamd_socket", + "is_active": True, + } + ) + self.assertEqual(backend.name, "Test Scanner") + self.assertEqual(backend.backend_type, "clamd_socket") + self.assertTrue(backend.is_active) + + def test_max_file_size_constraint(self): + """Test that max file size must be positive.""" + with self.assertRaises(ValidationError): + self.scanner_backend.create( + { + "name": "Test Scanner", + "backend_type": "clamd_socket", + "max_file_size_mb": 0, + } + ) + + def test_scan_timeout_constraint(self): + """Test that scan timeout must be positive.""" + with self.assertRaises(ValidationError): + self.scanner_backend.create( + { + "name": "Test Scanner", + "backend_type": "clamd_socket", + "scan_timeout_seconds": -1, + } + ) + + def test_clamd_port_constraint(self): + """Test that ClamAV port must be in valid range.""" + with self.assertRaises(ValidationError): + self.scanner_backend.create( + { + "name": "Test Scanner", + "backend_type": "clamd_network", + "clamd_port": 70000, # Invalid port + } + ) + + def test_get_active_scanner(self): + """Test getting active scanner backend.""" + # Create inactive backend (intentionally unused - verifies active scanner returns correct one) + self.scanner_backend.create( + { + "name": "Inactive Scanner", + "backend_type": "clamd_socket", + "is_active": False, + } + ) + + # Create active backend + active = self.scanner_backend.create( + { + "name": "Active Scanner", + "backend_type": "clamd_socket", + "is_active": True, + "sequence": 5, + } + ) + + # Should return the active scanner + result = self.scanner_backend.get_active_scanner() + self.assertEqual(result, active) + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_scan_binary_clean_file(self, mock_pyclamd): + """Test scanning a clean file.""" + # Mock ClamAV scanner + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = None # Clean file + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + backend = self.scanner_backend.create( + { + "name": "Test Scanner", + "backend_type": "clamd_socket", + "is_active": True, + } + ) + + result = backend.scan_binary(b"test data", filename="test.txt") + + self.assertEqual(result["status"], "clean") + self.assertIsNone(result["threat_name"]) + mock_scanner.scan_stream.assert_called_once() + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_scan_binary_infected_file(self, mock_pyclamd): + """Test scanning an infected file.""" + # Mock ClamAV scanner + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = {"stream": ("FOUND", "Eicar-Test-Signature")} + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + backend = self.scanner_backend.create( + { + "name": "Test Scanner", + "backend_type": "clamd_socket", + "is_active": True, + } + ) + + result = backend.scan_binary(b"test virus data", filename="virus.exe") + + self.assertEqual(result["status"], "infected") + self.assertEqual(result["threat_name"], "Eicar-Test-Signature") + + def test_scan_binary_file_too_large(self): + """Test skipping scan for files that are too large.""" + backend = self.scanner_backend.create( + { + "name": "Test Scanner", + "backend_type": "clamd_socket", + "is_active": True, + "max_file_size_mb": 1, # 1 MB limit + } + ) + + # Create 2 MB of data + large_data = b"x" * (2 * 1024 * 1024) + + result = backend.scan_binary(large_data) + + self.assertEqual(result["status"], "skipped") + self.assertIn("exceeds maximum", result["details"]) + + def test_scan_binary_inactive_backend(self): + """Test that inactive backends skip scanning.""" + backend = self.scanner_backend.create( + { + "name": "Test Scanner", + "backend_type": "clamd_socket", + "is_active": False, + } + ) + + result = backend.scan_binary(b"test data") + + self.assertEqual(result["status"], "skipped") + self.assertIn("not active", result["details"]) + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_test_connection_success(self, mock_pyclamd): + """Test successful connection test.""" + # Mock ClamAV scanner + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.version.return_value = "ClamAV 0.103.0" + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + backend = self.scanner_backend.create( + { + "name": "Test Scanner", + "backend_type": "clamd_socket", + } + ) + + result = backend.test_connection() + + self.assertTrue(backend.last_connection_test_result) + self.assertFalse(backend.last_connection_error) # Odoo Char fields are False when empty + self.assertEqual(result["type"], "ir.actions.client") + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_test_connection_failure(self, mock_pyclamd): + """Test failed connection test.""" + # Mock ClamAV scanner to raise exception + mock_pyclamd.ClamdUnixSocket.side_effect = Exception("Connection refused") + + backend = self.scanner_backend.create( + { + "name": "Test Scanner", + "backend_type": "clamd_socket", + } + ) + + result = backend.test_connection() + + self.assertFalse(backend.last_connection_test_result) + self.assertIsNotNone(backend.last_connection_error) + self.assertEqual(result["params"]["type"], "danger") + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_scan_binary_network_socket(self, mock_pyclamd): + """Test scanning using network socket backend type.""" + # Mock ClamAV network scanner + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = None # Clean file + mock_pyclamd.ClamdNetworkSocket.return_value = mock_scanner + + backend = self.scanner_backend.create( + { + "name": "Network Scanner", + "backend_type": "clamd_network", + "clamd_host": "192.168.1.100", + "clamd_port": 3310, + "is_active": True, + } + ) + + result = backend.scan_binary(b"test data", filename="test.txt") + + self.assertEqual(result["status"], "clean") + # Verify ClamdNetworkSocket was called with correct params + mock_pyclamd.ClamdNetworkSocket.assert_called_once_with( + host="192.168.1.100", + port=3310, + timeout=backend.scan_timeout_seconds, + ) + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_network_connection_test_success(self, mock_pyclamd): + """Test successful connection test with network socket.""" + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.version.return_value = "ClamAV 0.105.0" + mock_pyclamd.ClamdNetworkSocket.return_value = mock_scanner + + backend = self.scanner_backend.create( + { + "name": "Network Scanner", + "backend_type": "clamd_network", + "clamd_host": "clamav-server", + "clamd_port": 3310, + } + ) + + result = backend.test_connection() + + self.assertTrue(backend.last_connection_test_result) + self.assertEqual(result["type"], "ir.actions.client") + mock_pyclamd.ClamdNetworkSocket.assert_called_once() + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_scan_binary_infected_network(self, mock_pyclamd): + """Test detecting infected file via network socket.""" + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = {"stream": ("FOUND", "Win.Trojan.Generic")} + mock_pyclamd.ClamdNetworkSocket.return_value = mock_scanner + + backend = self.scanner_backend.create( + { + "name": "Network Scanner", + "backend_type": "clamd_network", + "is_active": True, + } + ) + + result = backend.scan_binary(b"malicious content") + + self.assertEqual(result["status"], "infected") + self.assertEqual(result["threat_name"], "Win.Trojan.Generic") diff --git a/spp_attachment_av_scan/tests/test_ir_attachment.py b/spp_attachment_av_scan/tests/test_ir_attachment.py new file mode 100644 index 00000000..a680ce96 --- /dev/null +++ b/spp_attachment_av_scan/tests/test_ir_attachment.py @@ -0,0 +1,730 @@ +import base64 +import logging +from unittest.mock import MagicMock, patch + +from odoo import Command, fields +from odoo.exceptions import AccessError, UserError +from odoo.tests import TransactionCase, tagged + +_logger = logging.getLogger(__name__) + + +@tagged("post_install", "-at_install") +class TestIrAttachment(TransactionCase): + """Test cases for ir.attachment with AV scanning.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.attachment_model = cls.env["ir.attachment"] + cls.scanner_backend_model = cls.env["spp.av.scanner.backend"] + + # Create a test scanner backend + cls.scanner_backend = cls.scanner_backend_model.create( + { + "name": "Test Scanner", + "backend_type": "clamd_socket", + "is_active": True, + } + ) + + def test_attachment_default_scan_status(self): + """Test that new attachments have pending scan status.""" + attachment = self.attachment_model.create( + { + "name": "test.txt", + "datas": base64.b64encode(b"test content"), + } + ) + + self.assertEqual(attachment.scan_status, "pending") + self.assertFalse(attachment.is_quarantined) + + def test_create_attachment_queues_scan(self): + """Test that creating binary attachment queues/processes scan.""" + # When job_worker is installed, the scan is queued automatically + # In test mode, the job may run synchronously + attachment = self.attachment_model.create( + { + "name": "test.txt", + "type": "binary", + "datas": base64.b64encode(b"test content"), + } + ) + + # The attachment should be created successfully + self.assertTrue(attachment.exists()) + # Status should be pending (queued) or already processed + self.assertIn(attachment.scan_status, ["pending", "clean", "error", "skipped"]) + + def test_scan_non_binary_attachment(self): + """Test that non-binary attachments are skipped.""" + attachment = self.attachment_model.create( + { + "name": "test.url", + "type": "url", + "url": "https://example.com", + } + ) + + attachment._scan_for_malware() + + self.assertEqual(attachment.scan_status, "skipped") + self.assertIn("Not a binary attachment", attachment.scan_result) + + def test_scan_no_active_backend(self): + """Test scanning when no active backend is configured.""" + # Deactivate all backends + self.scanner_backend.is_active = False + + attachment = self.attachment_model.create( + { + "name": "test.txt", + "type": "binary", + "datas": base64.b64encode(b"test content"), + } + ) + + attachment._scan_for_malware() + + self.assertEqual(attachment.scan_status, "skipped") + self.assertIn("No active scanner", attachment.scan_result) + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_scan_clean_file(self, mock_pyclamd): + """Test scanning a clean file.""" + # Mock ClamAV scanner + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = None # Clean + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + attachment = self.attachment_model.create( + { + "name": "clean.txt", + "type": "binary", + "datas": base64.b64encode(b"clean content"), + } + ) + + attachment._scan_for_malware() + + self.assertEqual(attachment.scan_status, "clean") + self.assertFalse(attachment.is_quarantined) + self.assertFalse(attachment.threat_name) # Odoo Char fields are False when empty + self.assertTrue(attachment.scan_date) + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_scan_infected_file(self, mock_pyclamd): + """Test scanning an infected file.""" + # Mock ClamAV scanner + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = {"stream": ("FOUND", "Eicar-Test-Signature")} + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + attachment = self.attachment_model.create( + { + "name": "virus.exe", + "type": "binary", + "datas": base64.b64encode(b"virus content"), + } + ) + + attachment._scan_for_malware() + + self.assertEqual(attachment.scan_status, "infected") + self.assertTrue(attachment.is_quarantined) + self.assertEqual(attachment.threat_name, "Eicar-Test-Signature") + self.assertIsNotNone(attachment.scan_date) + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_scan_error_handling(self, mock_pyclamd): + """Test error handling during scan.""" + # Mock ClamAV scanner to raise exception + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.side_effect = Exception("Scan failed") + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + attachment = self.attachment_model.create( + { + "name": "test.txt", + "type": "binary", + "datas": base64.b64encode(b"test content"), + } + ) + + attachment._scan_for_malware() + + self.assertEqual(attachment.scan_status, "error") + self.assertIn("error", attachment.scan_result.lower()) + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_quarantined_file_read_blocked(self, mock_pyclamd): + """Test that quarantined files cannot be read.""" + # Mock ClamAV scanner + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = {"stream": ("FOUND", "Eicar-Test-Signature")} + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + attachment = self.attachment_model.create( + { + "name": "virus.exe", + "type": "binary", + "datas": base64.b64encode(b"virus content"), + } + ) + + attachment._scan_for_malware() + + # Try to read the attachment + result = attachment.read(["datas"]) + + # Verify datas is blocked + self.assertFalse(result[0]["datas"]) + + def test_action_rescan(self): + """Test manual rescan action.""" + attachment = self.attachment_model.create( + { + "name": "test.txt", + "type": "binary", + "datas": base64.b64encode(b"test content"), + } + ) + # Set to clean to simulate previous scan + attachment.with_context(skip_av_scan_queue=True).write( + { + "scan_status": "clean", + } + ) + + result = attachment.action_rescan() + + # Verify scan was queued - status should be pending + self.assertEqual(attachment.scan_status, "pending") + self.assertEqual(result["type"], "ir.actions.client") + + def test_action_rescan_non_binary(self): + """Test that rescanning non-binary attachment raises error.""" + attachment = self.attachment_model.create( + { + "name": "test.url", + "type": "url", + "url": "https://example.com", + } + ) + + with self.assertRaises(UserError): + attachment.action_rescan() + + def test_write_datas_queues_rescan(self): + """Test that updating datas field queues rescan.""" + attachment = self.attachment_model.create( + { + "name": "test.txt", + "type": "binary", + "datas": base64.b64encode(b"original content"), + } + ) + # Set to clean to simulate previous scan + attachment.with_context(skip_av_scan_queue=True).write( + { + "scan_status": "clean", + } + ) + + # Update the datas field + attachment.write( + { + "datas": base64.b64encode(b"updated content"), + } + ) + + # Verify rescan was queued - status should be pending + self.assertEqual(attachment.scan_status, "pending") + + +@tagged("post_install", "-at_install") +class TestEncryptedQuarantine(TransactionCase): + """Test cases for encrypted quarantine functionality.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.attachment_model = cls.env["ir.attachment"] + cls.scanner_backend_model = cls.env["spp.av.scanner.backend"] + cls.encryption_provider_model = cls.env["spp.encryption.provider"] + + # Create a test scanner backend + cls.scanner_backend = cls.scanner_backend_model.create( + { + "name": "Test Scanner", + "backend_type": "clamd_socket", + "is_active": True, + } + ) + + # Create AV admin user for testing admin actions + cls.av_admin_group = cls.env.ref("spp_attachment_av_scan.group_av_admin") + cls.av_admin_user = cls.env["res.users"].create( + { + "name": "AV Admin", + "login": "av_admin_test", + "group_ids": [Command.link(cls.av_admin_group.id), Command.link(cls.env.ref("base.group_user").id)], + } + ) + + # Create regular user without AV admin access + cls.regular_user = cls.env["res.users"].create( + { + "name": "Regular User", + "login": "regular_user_test", + "group_ids": [Command.link(cls.env.ref("base.group_user").id)], + } + ) + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_quarantine_stores_hash(self, mock_pyclamd): + """Test that quarantine stores file hash correctly.""" + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = {"stream": ("FOUND", "Test-Virus")} + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + original_content = b"infected file content" + attachment = self.attachment_model.create( + { + "name": "virus.exe", + "type": "binary", + "datas": base64.b64encode(original_content), + } + ) + + attachment._scan_for_malware() + + self.assertTrue(attachment.is_quarantined) + self.assertTrue(attachment.quarantine_hash) + self.assertTrue(attachment.quarantine_date) + self.assertEqual(attachment.original_file_size, len(original_content)) + + # Verify hash is correct + import hashlib + + expected_hash = hashlib.sha256(original_content).hexdigest() + self.assertEqual(attachment.quarantine_hash, expected_hash) + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_quarantine_clears_original_datas(self, mock_pyclamd): + """Test that quarantine clears the original file data.""" + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = {"stream": ("FOUND", "Test-Virus")} + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + attachment = self.attachment_model.create( + { + "name": "virus.exe", + "type": "binary", + "datas": base64.b64encode(b"infected content"), + } + ) + + attachment._scan_for_malware() + + # Original datas should be cleared + self.assertFalse(attachment.datas) + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_restore_quarantined_admin_only(self, mock_pyclamd): + """Test that only AV admins can restore quarantined files.""" + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = {"stream": ("FOUND", "Test-Virus")} + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + attachment = self.attachment_model.create( + { + "name": "virus.exe", + "type": "binary", + "datas": base64.b64encode(b"infected content"), + } + ) + + attachment._scan_for_malware() + + # Regular user should not be able to restore + with self.assertRaises(AccessError): + attachment.with_user(self.regular_user).action_restore_quarantined() + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_permanently_delete_quarantined(self, mock_pyclamd): + """Test permanently deleting a quarantined file.""" + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = {"stream": ("FOUND", "Test-Virus")} + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + attachment = self.attachment_model.create( + { + "name": "virus.exe", + "type": "binary", + "datas": base64.b64encode(b"infected content"), + } + ) + + attachment._scan_for_malware() + attachment_id = attachment.id + + # Delete as admin + attachment.with_user(self.av_admin_user).action_permanently_delete_quarantined() + + # Attachment should no longer exist + self.assertFalse(self.attachment_model.browse(attachment_id).exists()) + + def test_action_rescan_quarantined_blocked(self): + """Test that rescanning a quarantined file is blocked.""" + attachment = self.attachment_model.create( + { + "name": "test.txt", + "type": "binary", + "datas": base64.b64encode(b"test content"), + } + ) + # Simulate quarantined state + attachment.with_context(skip_av_scan_queue=True).write( + { + "is_quarantined": True, + "scan_status": "infected", + } + ) + + with self.assertRaises(UserError): + attachment.action_rescan() + + def test_cron_purge_old_quarantined(self): + """Test scheduled purge of old quarantined files.""" + from datetime import timedelta + + attachment = self.attachment_model.create( + { + "name": "old_virus.exe", + "type": "binary", + "datas": base64.b64encode(b"test"), + } + ) + + # Simulate old quarantine + old_date = fields.Datetime.now() - timedelta(days=100) + attachment.with_context(skip_av_scan_queue=True).write( + { + "is_quarantined": True, + "quarantine_date": old_date, + "quarantine_data": base64.b64encode(b"encrypted_data"), + "scan_status": "infected", + } + ) + + # Set retention to 90 days + self.env["ir.config_parameter"].sudo().set_param("spp_attachment_av_scan.quarantine_retention_days", "90") + + # Run purge + self.attachment_model._cron_purge_old_quarantined_files() + + # Quarantine data should be cleared + attachment.invalidate_recordset() + self.assertFalse(attachment.quarantine_data) + + def test_calculate_file_hash(self): + """Test file hash calculation.""" + attachment = self.attachment_model.create( + { + "name": "test.txt", + "type": "binary", + "datas": base64.b64encode(b"test content"), + } + ) + + test_data = b"test content for hashing" + calculated_hash = attachment._calculate_file_hash(test_data) + + import hashlib + + expected_hash = hashlib.sha256(test_data).hexdigest() + self.assertEqual(calculated_hash, expected_hash) + + def test_restore_not_quarantined_error(self): + """Test that restoring non-quarantined file raises error.""" + attachment = self.attachment_model.create( + { + "name": "clean.txt", + "type": "binary", + "datas": base64.b64encode(b"clean content"), + } + ) + attachment.with_context(skip_av_scan_queue=True).write( + { + "scan_status": "clean", + "is_quarantined": False, + } + ) + + with self.assertRaises(UserError): + attachment.with_user(self.av_admin_user).action_restore_quarantined() + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_notification_sent_on_infection(self, mock_pyclamd): + """Test that security admins are notified when malware is detected.""" + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = {"stream": ("FOUND", "Test-Malware")} + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + # Ensure we have the AV admin group with users + av_admin_group = self.env.ref("spp_attachment_av_scan.group_av_admin") + self.assertTrue(av_admin_group.user_ids, "AV admin group should have users") + + attachment = self.attachment_model.create( + { + "name": "malware.exe", + "type": "binary", + "datas": base64.b64encode(b"malicious content"), + } + ) + + # Count messages before scan + messages_before = self.env["mail.message"].search_count( + [("model", "=", "ir.attachment"), ("res_id", "=", attachment.id)] + ) + + attachment._scan_for_malware() + + # Count messages after scan + messages_after = self.env["mail.message"].search_count( + [("model", "=", "ir.attachment"), ("res_id", "=", attachment.id)] + ) + + # A notification message should have been created + self.assertGreater(messages_after, messages_before) + + def test_forensic_download_marked(self): + """Test that forensic downloads are marked with is_forensic_download flag.""" + attachment = self.attachment_model.create( + { + "name": "test.txt", + "type": "binary", + "datas": base64.b64encode(b"test"), + } + ) + + # Simulate quarantine state with encrypted data + # First create an encryption provider with a key + encryption_provider = self.env["spp.encryption.provider"].sudo().search([("type", "=", "jwcrypto")], limit=1) + + if not encryption_provider: + _logger.info("Skipping test_forensic_download_marked: no encryption provider available") + return + + try: + original_content = b"quarantined content" + encrypted_data = encryption_provider.encrypt_data(original_content) + file_hash = attachment._calculate_file_hash(original_content) + + attachment.with_context(skip_av_scan_queue=True).write( + { + "is_quarantined": True, + "scan_status": "infected", + "quarantine_data": base64.b64encode(encrypted_data), + "quarantine_hash": file_hash, + "original_file_size": len(original_content), + "datas": False, + } + ) + + # Download for analysis as admin + attachment.with_user(self.av_admin_user).action_download_quarantined_for_analysis() + + # Find the created download attachment + download_attachment = self.attachment_model.search([("name", "=", f"QUARANTINED_{attachment.name}")]) + + self.assertTrue(download_attachment.exists()) + self.assertTrue(download_attachment.is_forensic_download) + self.assertEqual(download_attachment.scan_status, "skipped") + except ValueError as e: + if "No encryption key" in str(e): + _logger.info("Skipping test_forensic_download_marked: %s", e) + else: + raise + + def test_cron_cleanup_forensic_downloads(self): + """Test scheduled cleanup of old forensic download attachments.""" + from datetime import timedelta + + # Create a forensic download attachment + attachment = self.attachment_model.create( + { + "name": "QUARANTINED_old_file.exe", + "type": "binary", + "datas": base64.b64encode(b"test"), + "is_forensic_download": True, + } + ) + + # Simulate old create_date (25 hours ago) + old_date = fields.Datetime.now() - timedelta(hours=25) + self.env.cr.execute( + "UPDATE ir_attachment SET create_date = %s WHERE id = %s", + (old_date, attachment.id), + ) + attachment.invalidate_recordset() + + # Set retention to 24 hours + self.env["ir.config_parameter"].sudo().set_param( + "spp_attachment_av_scan.forensic_download_retention_hours", "24" + ) + + attachment_id = attachment.id + + # Run cleanup + self.attachment_model._cron_cleanup_forensic_downloads() + + # Attachment should be deleted + self.assertFalse(self.attachment_model.browse(attachment_id).exists()) + + def test_cron_cleanup_keeps_recent_forensic_downloads(self): + """Test that recent forensic downloads are not cleaned up.""" + # Create a recent forensic download attachment + attachment = self.attachment_model.create( + { + "name": "QUARANTINED_recent_file.exe", + "type": "binary", + "datas": base64.b64encode(b"test"), + "is_forensic_download": True, + } + ) + + # Set retention to 24 hours + self.env["ir.config_parameter"].sudo().set_param( + "spp_attachment_av_scan.forensic_download_retention_hours", "24" + ) + + attachment_id = attachment.id + + # Run cleanup + self.attachment_model._cron_cleanup_forensic_downloads() + + # Attachment should still exist (it's recent) + self.assertTrue(self.attachment_model.browse(attachment_id).exists()) + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_large_file_skipped_during_scan(self, mock_pyclamd): + """Test that files exceeding max_file_size_mb are skipped.""" + # Configure scanner with 1 MB limit + self.scanner_backend.write({"max_file_size_mb": 1}) + + # Create attachment with 2 MB of data + large_content = b"x" * (2 * 1024 * 1024) + attachment = self.attachment_model.create( + { + "name": "large_file.bin", + "type": "binary", + "datas": base64.b64encode(large_content), + } + ) + + # Run scan + attachment._scan_for_malware() + + # Should be skipped due to size + self.assertEqual(attachment.scan_status, "skipped") + self.assertIn("exceeds maximum", attachment.scan_result) + + def test_size_validation_on_restore(self): + """Test that size validation catches corrupted restored files.""" + attachment = self.attachment_model.create( + { + "name": "test.txt", + "type": "binary", + "datas": base64.b64encode(b"test"), + } + ) + + encryption_provider = self.env["spp.encryption.provider"].sudo().search([("type", "=", "jwcrypto")], limit=1) + + if not encryption_provider: + _logger.info("Skipping test_size_validation_on_restore: no encryption provider available") + return + + try: + original_content = b"original content here" + # Encrypt different data than what original_file_size indicates + different_content = b"different" + encrypted_data = encryption_provider.encrypt_data(different_content) + file_hash = attachment._calculate_file_hash(different_content) + + attachment.with_context(skip_av_scan_queue=True).write( + { + "is_quarantined": True, + "scan_status": "infected", + "quarantine_data": base64.b64encode(encrypted_data), + "quarantine_hash": file_hash, + # Set wrong size to simulate corruption + "original_file_size": len(original_content), + "datas": False, + } + ) + + # Restore should fail due to size mismatch + with self.assertRaises(UserError) as context: + attachment.with_user(self.av_admin_user).action_restore_quarantined() + + self.assertIn("size mismatch", str(context.exception).lower()) + except ValueError as e: + if "No encryption key" in str(e): + _logger.info("Skipping test_size_validation_on_restore: %s", e) + else: + raise + + @patch("odoo.addons.spp_attachment_av_scan.models.av_scanner_backend.pyclamd") + def test_full_restore_flow_with_encryption(self, mock_pyclamd): + """Test complete quarantine and restore flow with encryption.""" + mock_scanner = MagicMock() + mock_scanner.ping.return_value = True + mock_scanner.scan_stream.return_value = {"stream": ("FOUND", "Test-Virus")} + mock_pyclamd.ClamdUnixSocket.return_value = mock_scanner + + original_content = b"this was falsely detected as malware" + attachment = self.attachment_model.create( + { + "name": "false_positive.doc", + "type": "binary", + "datas": base64.b64encode(original_content), + } + ) + + # Scan and quarantine + attachment._scan_for_malware() + + self.assertTrue(attachment.is_quarantined) + self.assertEqual(attachment.scan_status, "infected") + self.assertFalse(attachment.datas) # Original data cleared + self.assertTrue(attachment.quarantine_hash) + self.assertEqual(attachment.original_file_size, len(original_content)) + + # Check if encryption provider exists for restore test + encryption_provider = self.env["spp.encryption.provider"].sudo().search([("type", "=", "jwcrypto")], limit=1) + + if encryption_provider and attachment.quarantine_data: + # Restore as admin + attachment.with_user(self.av_admin_user).action_restore_quarantined() + + # Verify restoration + self.assertFalse(attachment.is_quarantined) + self.assertEqual(attachment.scan_status, "clean") + self.assertTrue(attachment.datas) + + # Verify restored content matches original + restored_content = base64.b64decode(attachment.datas) + self.assertEqual(restored_content, original_content) diff --git a/spp_attachment_av_scan/views/av_scanner_backend_views.xml b/spp_attachment_av_scan/views/av_scanner_backend_views.xml new file mode 100644 index 00000000..db4714c6 --- /dev/null +++ b/spp_attachment_av_scan/views/av_scanner_backend_views.xml @@ -0,0 +1,106 @@ + + + + + spp.av.scanner.backend.tree + spp.av.scanner.backend + + + + + + + + + + + + + + + spp.av.scanner.backend.form + spp.av.scanner.backend + +
+
+
+ +
+ +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
+ + + + + + Antivirus Scanner Backends + spp.av.scanner.backend + list,form + +

+ Create your first antivirus scanner backend +

+

+ Configure ClamAV or other antivirus scanners to automatically + scan uploaded files for malware. +

+ + + + + + + diff --git a/spp_attachment_av_scan/views/ir_attachment_views.xml b/spp_attachment_av_scan/views/ir_attachment_views.xml new file mode 100644 index 00000000..d2d29406 --- /dev/null +++ b/spp_attachment_av_scan/views/ir_attachment_views.xml @@ -0,0 +1,145 @@ + + + + + ir.attachment.form.inherit.av.scan + ir.attachment + + + + + + + + + + + + + + + + +
+

+ +

+
+ + +
+
+