Skip to content

Add strict TOTP validation to block invalid authentication requests #578

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sharmagot merged 12 commits into from
Jan 9, 2026
Merged

Add strict TOTP validation to block invalid authentication requests #578

Changes from 3 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
f5a2d64
Added explicit validation to reject TOTP values containing non-numeri…
sharmagot Dec 8, 2025
67c50e7
Improve TOTP validation: trim whitespace and standardize auth error logs
sharmagot Dec 9, 2025
d9ac515
Use shared TOTP validator for normalization and consistent error mess…
sharmagot Dec 14, 2025
65f5699
Fix TOTP validation and block invalid authentication requests
sharmagot Jan 2, 2026
a8c0e4b
Unify TOTP validation to return a single generic error message
sharmagot Jan 2, 2026
fb1f0e8
refactor: unify TOTP authentication error handling
sharmagot Jan 4, 2026
50600fb
Normalize TOTP validation and unify invalid TOTP error handling
sharmagot Jan 6, 2026
ea83763
test_totp_invalid_alphanumeric_code
sharmagot Jan 6, 2026
2ff4cdf
corrected the test case
sharmagot Jan 6, 2026
fc79ca2
corrected the test case
sharmagot Jan 8, 2026
8441867
fixed the test case failure
sharmagot Jan 8, 2026
c1ed4ca
removed the unnecessary test case
sharmagot Jan 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 54 additions & 13 deletions vertica_python/vertica/connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -313,6 +313,31 @@ def __init__(self, options: Optional[Dict[str, Any]] = None) -> None:
if self.totp is not None:
if not isinstance(self.totp, str):
raise TypeError('The value of connection option "totp" should be a string')
# Use shared TOTP validator for normalization and precedence checks
try:
from .totp_validation import validate_totp_code, INVALID_TOTP_MSG
except Exception:
validate_totp_code = None
INVALID_TOTP_MSG = 'Invalid TOTP: Please enter a valid 6-digit numeric code.'

if validate_totp_code is not None:
result = validate_totp_code(self.totp, totp_is_valid=None)
Copy link
Collaborator

@sivaalamp sivaalamp Jan 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

validate_totp_code function definition is missing.

Copy link
Collaborator Author

@sharmagot sharmagot Jan 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi Siva , Added the definition for validate_totp_code.

if not result.ok:
msg = result.message or INVALID_TOTP_MSG
self._logger.error(f'Authentication failed: {msg}')
raise errors.ConnectionError(f'Authentication failed: {msg}')
# normalized digits-only code
self.totp = result.code
else:
# Fallback minimal validation
s = self.totp.strip()
if not s.isdigit():
self._logger.error(INVALID_TOTP_MSG)
raise errors.ConnectionError(INVALID_TOTP_MSG)
if len(s) != 6:
self._logger.error(INVALID_TOTP_MSG)
raise errors.ConnectionError(INVALID_TOTP_MSG)
self.totp = s
self._logger.info('TOTP received in connection options')

# OAuth authentication setup
Expand Down Expand Up @@ -974,10 +999,14 @@ def send_startup(totp_value=None):
short_msg = match.group(1).strip() if match else error_msg.strip()

if "Invalid TOTP" in short_msg:
print("Authentication failed: Invalid TOTP token.")
self._logger.error("Authentication failed: Invalid TOTP token.")
try:
from .totp_validation import INVALID_TOTP_MSG
except Exception:
INVALID_TOTP_MSG = "Invalid TOTP: Please enter a valid 6-digit numeric code."
print(f"Authentication failed: {INVALID_TOTP_MSG}")
self._logger.error(f"Authentication failed: {INVALID_TOTP_MSG}")
self.close_socket()
raise errors.ConnectionError("Authentication failed: Invalid TOTP token.")
raise errors.ConnectionError(f"Authentication failed: {INVALID_TOTP_MSG}")

# Generic error fallback
print(f"Authentication failed: {short_msg}")
Expand All @@ -1000,16 +1029,28 @@ def send_startup(totp_value=None):
if ready:
Copy link
Collaborator

@sivaalamp sivaalamp Jan 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we wait 5mins for the user to enter TOTP?

Copy link
Collaborator Author

@sharmagot sharmagot Jan 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi Siva, if the user does not provide the TOTP within 5 minutes, the session should expire—meaning the connection between the driver and the server should be terminated.

totp_input = sys.stdin.readline().strip()

# ❌ Blank TOTP entered
if not totp_input:
self._logger.error("Invalid TOTP: Cannot be empty.")
raise errors.ConnectionError("Invalid TOTP: Cannot be empty.")

# ❌ Validate TOTP format (must be 6 digits)
if not totp_input.isdigit() or len(totp_input) != 6:
print("Invalid TOTP format. Please enter a 6-digit code.")
self._logger.error("Invalid TOTP format entered.")
raise errors.ConnectionError("Invalid TOTP format: Must be a 6-digit number.")
# Validate using shared precedence
try:
from .totp_validation import validate_totp_code, INVALID_TOTP_MSG
except Exception:
validate_totp_code = None
INVALID_TOTP_MSG = "Invalid TOTP: Please enter a valid 6-digit numeric code."

if validate_totp_code is not None:
result = validate_totp_code(totp_input, totp_is_valid=None)
if not result.ok:
msg = result.message or INVALID_TOTP_MSG
print(msg)
Copy link
Collaborator

@sivaalamp sivaalamp Jan 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is this print(msg) for debugging purpose, please remove it

Copy link
Collaborator Author

@sharmagot sharmagot Jan 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi Siva, I have removed the debug message .

self._logger.error(msg)
raise errors.ConnectionError(msg)
totp_input = result.code
else:
s = totp_input.strip()
if not s.isdigit() or len(s) != 6:
print(INVALID_TOTP_MSG)
self._logger.error(INVALID_TOTP_MSG)
raise errors.ConnectionError(INVALID_TOTP_MSG)
totp_input = s
# ✅ Valid TOTP — retry connection
totp = totp_input
self.close_socket()
Expand Down