End-to-end CVE history support and hardens loading/startup

- Added end-to-end cvehist support: DB table/migration, models, loader, CLI search, and /api/search/cvehist.
- Extended search filters for CVE history: CVE ID, change dates, change-event, and change-type.
- Hardened loading: upserts for CVE/CPE/history, timeouts, resume support, and cleaner EPSS sync/error handling.
- Improved startup: DB schema migration now runs automatically before the web app starts.
- Updated config/docs for the CVE history API and new load/search commands.

Author: Vadim Bogulean

Dockerfile

@@ -19,4 +19,4 @@ COPY ./src ${FCDB_HOME}
 
 EXPOSE 8000
 
-CMD ["uvicorn", "web.app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
+CMD ["/bin/sh", "-c", "python -m web.prestart && exec uvicorn web.app:app ${FCDB_WEB_PARAMS:---host 0.0.0.0 --port 8000 --workers 4}"]

README.md

@@ -147,17 +147,17 @@ Restore:
 docker compose exec -T fastcve-db sh -c 'pg_restore -U "$POSTGRES_USER" -d "$POSTGRES_DB" --clean --if-exists "/backup/fastcve_vuln_db_YYYY-MM-DD.dump"'
 ```
 
-- **Populate the DB for the first time (CVE/CPE/CWE/CAPEC/EPSS/KEV)**:
+- **Populate the DB for the first time (CVE/CVE history/CPE/CWE/CAPEC/EPSS/KEV)**:
 ```bash
-docker compose exec fastcve load --data cve cpe cwe capec epss kev
+docker compose exec fastcve load --data cve cvehist cpe cwe capec epss kev
 ```
 
-- **Update CVE/CPE incrementally (and refresh EPSS/KEV)**:
+- **Update CVE/CVE history/CPE incrementally (and refresh EPSS/KEV)**:
 ```bash
-docker compose exec fastcve load --data cve cpe epss kev
+docker compose exec fastcve load --data cve cvehist cpe epss kev
 ```
 
-This fetches changes since the last successful update (for CVE/CPE), with an upper limit of `fetch.max.days.period` (default 120 days) enforced by the loader.
+This fetches changes since the last successful update (for CVE/CVE history/CPE), with an upper limit of `fetch.max.days.period` (default 120 days) enforced by the loader.
 
 If there is a need to repopulate the DB for the CWE/CAPEC info, then `--full` and `--drop` options are available for the load command. `--full` ignores the fact the data is already present and `--drop` drops existing data before loading. When using `--data epss` in combination with `--epss-now`, the loader fetches EPSS data for the current date; otherwise it defaults to the previous day.
 
@@ -195,6 +195,14 @@ Additional filters are available for CVE search:
 --last-mod-start-date # retrieve only those CVEs that are last modified after the start date
 --last-mod-end-date   # retrieve only those CVEs that are last modified before the end date
 ```
+
+- search for the data: **get CVE history rows for a CVE or change pattern**
+```
+docker compose exec fastcve search --search-info cvehist --cve CVE-2024-3094 --change-event 'Initial Analysis' --output json
+```
+
+Above will return the matching CVE history change records. `cvehist` supports filtering by CVE ID, change date range, `--change-event`, and `--change-type`.
+
 - search for the data: **get the valid list of CPE names for a query on part/vendor/product/version etc**.
 
 ```
@@ -216,6 +224,7 @@ The following endpoints are exposed through HTTP requests
 ```
 /status - DB status
 /api/search/cve - search for CVE data
+/api/search/cvehist - search for CVE history data
 /api/search/cpe - search for CPE data
 /api/search/cwe - search for CWE data
 /api/search/capec - search for CAPEC data

src/common/models/cve_history.py

@@ -0,0 +1,25 @@
+from datetime import datetime
+from typing import List, Optional, Any
+from pydantic import BaseModel
+
+
+class CveChangeDetail(BaseModel):
+    class Config:
+        extra = 'allow'  # details are not fully stable
+
+    type: Optional[str] = None
+    oldValue: Optional[Any] = None
+    newValue: Optional[Any] = None
+
+
+
+class CveHistoryItem(BaseModel):
+    class Config:
+        extra = 'ignore'
+
+    cveId: str
+    cveChangeId: str
+    eventName: str
+    sourceIdentifier: Optional[str] = None
+    created: datetime
+    details: Optional[List[CveChangeDetail]] = None

src/common/models/models.py

@@ -75,6 +75,8 @@ class SearchInfoType(str, Enum):
     cpe = "cpe"
     capec = "capec"
     status = "status"
+    cvehist = "cvehist"
+
 
 
 class CveSeverityV2(str, Enum):
@@ -129,6 +131,8 @@ class SearchOptions(BaseModel):
     lastModEndDate: Optional[date] = Field(default=None, description="Last modified end date", alias="last-mod-end-date")
     pubStartDate: Optional[date] = Field(default=None, description="CVE Published start date", alias="pub-start-date")
     pubEndDate: Optional[date] = Field(default=None, description="CVE Published start date", alias="pub-end-date")
+    changeEvent: Optional[str] = Field(default=None, description="Regexp to filter CVE history by change event name (e.g. 'CVE Modified', 'Initial Analysis')", cli=('--change-event',),alias="change-event")
+    changeType: Optional[str] = Field(default=None, description="Regexp to filter CVE history by change detail type (e.g. 'Reference', 'CVSS V3.1')", cli=('--change-type',), alias="change-type")
     vulnerable: Optional[bool] = Field(default=True, description="CVE found by the CPEs that are marked as vulnerable", alias="vulnerable")
     pageSize: Optional[int] = Field(description="Number of results per page", default=100, alias="page-size", ge=10, le=3000)
     pageIdx: Optional[int] = Field(default=0, description="Starting index", alias="page-idx", ge=0)
@@ -153,8 +157,8 @@ def validate_cpe_name(cls, value):
     def validate_mandatory_fields(cls, inst):
         """Implement the root validator"""
 
-        # Validate input parameters in case of search-info set as CVE
-        if inst.get('searchInfo', None) in (SearchInfoType.cve, SearchInfoType.cpe):
+        # Validate input parameters in case of search-info set as CVE/CPE/CveHistory
+        if inst.get('searchInfo', None) in (SearchInfoType.cve, SearchInfoType.cpe, SearchInfoType.cvehist):
             if inst['lastModStartDate'] and inst['lastModEndDate'] and inst['lastModStartDate'] > inst['lastModEndDate']:
                 exc = ValueError('Last modified start date must be before last modified end date')
                 raise ValidationError([ErrorWrapper(exc, loc=cls.__fields__['lastModStartDate'].alias)], cls)

src/common/search.py

@@ -11,15 +11,27 @@
 
 import re
 import json
+from functools import lru_cache
 from typing import List, Iterator
 from sqlalchemy import Boolean,  cast, Numeric, select
 from sqlalchemy.sql import text, expression
 from sqlalchemy.orm import aliased
 from generic import ApplicationContext
-from db.tables import Vuln, VulnCpes, Cpe, Cwe, FetchStatus, Capec
+from db.tables import Vuln, VulnCpes, Cpe, Cwe, FetchStatus, Capec, VulnHistory
 from common.models import SearchOptions, SearchInfoType, OutputType, CPE23_REGEX_STR, metrics_mapping
+from datetime import datetime, time, timezone
 
 class ValidationError(Exception): ...
+class MissingSearchDataError(ValidationError): ...
+
+
+SEARCH_INFO_DATASETS = {
+    SearchInfoType.cve: dict(model=Vuln, label="CVE", load_arg="cve"),
+    SearchInfoType.cpe: dict(model=Cpe, label="CPE", load_arg="cpe"),
+    SearchInfoType.cwe: dict(model=Cwe, label="CWE", load_arg="cwe"),
+    SearchInfoType.capec: dict(model=Capec, label="CAPEC", load_arg="capec"),
+    SearchInfoType.cvehist: dict(model=VulnHistory, label="CVE history", load_arg="cvehist"),
+}
 
 # regex used to split the cpe 2.3 into separate pieces
 COLUMN_REGEX = re.compile(r'(?<!\\):')
@@ -30,6 +42,41 @@ def get_non_empty_opts(opts: SearchOptions) -> dict:
     return {k: v for k, v in vars(opts).items() if v is not None}
 
 
+# ------------------------------------------------------------------------------
+def get_missing_search_data_message(search_info: SearchInfoType) -> str:
+
+    dataset = SEARCH_INFO_DATASETS[search_info]
+    return (
+        f"No {dataset['label']} data is loaded in the database for search-info={search_info.value}. "
+        f"Load it first with: load --data {dataset['load_arg']}"
+    )
+
+
+# ------------------------------------------------------------------------------
+def get_search_data_cache_bucket() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%d")
+
+
+# ------------------------------------------------------------------------------
+@lru_cache(maxsize=len(SEARCH_INFO_DATASETS))
+def _ensure_search_data_loaded_cached(search_info: SearchInfoType, cache_bucket: str) -> SearchInfoType:
+
+    dataset = SEARCH_INFO_DATASETS.get(search_info)
+    if not dataset:
+        return search_info
+
+    with ApplicationContext.instance().db as session:
+        row = session.execute(select(dataset['model'].id).limit(1)).first()
+        if row is None:
+            raise MissingSearchDataError(get_missing_search_data_message(search_info))
+    return search_info
+
+
+# ------------------------------------------------------------------------------
+def ensure_search_data_loaded(appctx: ApplicationContext, search_info: SearchInfoType) -> None:
+    _ensure_search_data_loaded_cached(search_info, get_search_data_cache_bucket())
+
+
 # ------------------------------------------------------------------------------
 def search_cves(appctx: ApplicationContext, opts: SearchOptions):
 
@@ -135,6 +182,79 @@ def search_cves(appctx: ApplicationContext, opts: SearchOptions):
 
     return result
 
+# ------------------------------------------------------------------------------
+def search_cvehist(appctx: ApplicationContext, opts: SearchOptions):
+
+    result = {}
+    vh_table = aliased(VulnHistory, name='vh_table')
+
+    with appctx.db as session:
+
+        filters = []
+
+        # filter by CVE IDs
+        if opts.cveId:
+            cve_ids = list({cve_id.upper() for cve_id in opts.cveId})
+            filters.append(vh_table.vuln_id.in_(cve_ids))
+
+        # filter by change event name (regex, case-insensitive)
+        if opts.changeEvent:
+            filters.append(vh_table.event_name.op("~*")(opts.changeEvent))
+
+        # filter by change type inside details[].type (JSONB)
+        if opts.changeType:
+            filters.append(
+                text(
+                    "EXISTS ("
+                    "SELECT 1 "
+                    "FROM jsonb_array_elements(COALESCE(vh_table.data->'details', '[]'::jsonb)) AS detail "
+                    "WHERE detail->>'type' ~* :ct"
+                    ")"
+                ).bindparams(ct=opts.changeType)
+            )
+
+        # filter by change date range
+        if opts.lastModStartDate:
+            start_dt = datetime.combine(opts.lastModStartDate, time.min)
+            filters.append(vh_table.change_date >= start_dt)
+        if opts.lastModEndDate:
+            end_dt = datetime.combine(opts.lastModEndDate, time.max)
+            filters.append(vh_table.change_date <= end_dt)
+
+        filtered_history_query = select(
+            vh_table.vuln_id.label("vuln_id"),
+            vh_table.change_date.label("change_date"),
+            vh_table.data.label("data"),
+        )
+        if filters:
+            filtered_history_query = filtered_history_query.where(*filters)
+        filtered_history = filtered_history_query.cte("filtered_history").prefix_with("MATERIALIZED")
+
+        # 1) paginate matching CVE IDs from the filtered rowset
+        cve_id_subquery = (
+            select(filtered_history.c.vuln_id)
+            .distinct()
+            .order_by(filtered_history.c.vuln_id)
+            .offset(opts.pageIdx * opts.pageSize)
+            .limit(opts.pageSize)
+            .subquery()
+        )
+
+        # 2) return the matching history rows for those paginated CVEs
+        query = (
+            select(filtered_history.c.data)
+            .where(filtered_history.c.vuln_id.in_(select(cve_id_subquery.c.vuln_id)))
+            .order_by(filtered_history.c.vuln_id, filtered_history.c.change_date)
+        )
+
+        rows = session.execute(query).all()
+
+        result = {
+            "search": get_non_empty_opts(opts),
+            "result": [row.data for row in rows],
+        }
+
+    return result
 
 # ------------------------------------------------------------------------------
 def get_cvss_metric_conditions(cvss_metrics: str, version:str) -> Iterator[dict]:
@@ -426,13 +546,15 @@ def search_capec(appctx: ApplicationContext, opts: SearchOptions):
 def search_data(appctx, opts: SearchOptions):
 
     search_results = {}
+    ensure_search_data_loaded(appctx, opts.searchInfo)
 
     # search the data based on the input criterias
     if   opts.searchInfo == SearchInfoType.status:  search_results = get_fetch_status(appctx)
     elif opts.searchInfo == SearchInfoType.cve:     search_results = search_cves(appctx, opts)
     elif opts.searchInfo == SearchInfoType.cpe:     search_results = search_cpes(appctx, opts)
     elif opts.searchInfo == SearchInfoType.cwe:     search_results = search_cwes(appctx, opts)
     elif opts.searchInfo == SearchInfoType.capec:   search_results = search_capec(appctx, opts)
+    elif opts.searchInfo == SearchInfoType.cvehist:        search_results = search_cvehist(appctx, opts)
 
     return search_results
 
@@ -453,6 +575,7 @@ def results_output_id(opts: SearchOptions, search_results):
         SearchInfoType.cpe:     'cpeName',
         SearchInfoType.cwe:     'ID',
         SearchInfoType.capec:   'ID',
+        SearchInfoType.cvehist: 'cveId',
     }
 
     key = key_names_map[opts.searchInfo]

src/common/util.py

@@ -6,6 +6,7 @@
 
 import os
 import shutil
+import time
 from alembic.config import Config
 from alembic import command
 
@@ -48,6 +49,26 @@ def init_db_schema():
         os.chdir(cwd)
 
 
+# ------------------------------------------------------------------------------
+def ensure_db_schema(max_attempts=60, retry_delay=1):
+
+    setup_env()
+
+    last_exc = None
+    for attempt in range(max_attempts):
+        try:
+            init_db_schema()
+            return
+        except Exception as exc:  # pragma: no cover
+            last_exc = exc
+            if attempt == max_attempts - 1:
+                break
+            time.sleep(retry_delay)
+
+    if last_exc is not None:  # pragma: no cover
+        raise last_exc
+
+
 # ------------------------------------------------------------------------------
 def setup_env():
 

src/config/setenv/config.ini

@@ -43,6 +43,9 @@ file.max.count = 10
 ; NIST CVE API
 url.cve = https://services.nvd.nist.gov/rest/json/cves/2.0
 
+; NIST CVE History API
+url.cvehist = https://services.nvd.nist.gov/rest/json/cvehistory/2.0
+
 ; NIST CPE API
 url.cpe = https://services.nvd.nist.gov/rest/json/cpes/2.0
 
@@ -67,6 +70,7 @@ api_key = ${NVD_API_KEY}
 ; pause between requests
 request.pause.with_key = 1 #seconds to pause between requests
 request.pause.without_key = 6 #seconds to pause between requests
+worker.timeout = 300 #seconds without worker progress before failing
 
 ; min time between syncs (sec)
 min.sync.time = 2 * 60 * 60 # sec

src/db/scripts/versions/91d0157eaab5_add_vuln_history_table.py

@@ -0,0 +1,56 @@
+"""add vuln_history table
+
+Revision ID: 91d0157eaab5
+Revises: ecd29e77afe3
+Create Date: 2026-01-27 12:48:44.685028
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision = '91d0157eaab5'
+down_revision = 'ecd29e77afe3'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "vuln_history",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("vuln_id", sa.String(length=20), nullable=False),
+        sa.Column("change_id", sa.String(length=64), nullable=False),
+        sa.Column("event_name", sa.String(length=64)),
+        sa.Column("source", sa.String(length=100)),
+        sa.Column("change_date", sa.DateTime(timezone=True)),
+        sa.Column("data", postgresql.JSONB(), nullable=False),
+        sa.Column(
+            "sys_creation_date",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+            nullable=False,
+        ),
+    )
+
+    op.create_index("ix_vuln_history_vuln_id", "vuln_history", ["vuln_id"])
+    op.create_index("ix_vuln_history_change_id", "vuln_history", ["change_id"], unique=True)
+    op.create_index("ix_vuln_history_event_name", "vuln_history", ["event_name"])
+    op.create_index("ix_vuln_history_change_date", "vuln_history", ["change_date"])
+
+    op.create_index(
+        "ix_vuln_history_data_gin",
+        "vuln_history",
+        ["data"],
+        postgresql_using="gin",
+    )
+
+def downgrade():
+    op.drop_index("ix_vuln_history_data_gin", table_name="vuln_history")
+    op.drop_index("ix_vuln_history_change_date", table_name="vuln_history")
+    op.drop_index("ix_vuln_history_event_name", table_name="vuln_history")
+    op.drop_index("ix_vuln_history_change_id", table_name="vuln_history")
+    op.drop_index("ix_vuln_history_vuln_id", table_name="vuln_history")
+    op.drop_table("vuln_history")